feat(events): identify the shared Redis ID space, behind a two-phase flip (BUG-2736)

The activity event counter lives in Redis and is shared by every instance, so
no instance can compute an identity for it the way MemoryBus computes its own
incarnation base. If that counter is ever reset -- evicted under maxmemory,
deleted by hand, a fresh Redis after a restore -- IDs start again from 1, and
a replica buffering the old sequence cannot tell the new 101 from the old 101.
It merges two ID spaces into one replay buffer and answers a resume across the
boundary as though nothing was missed.

Numeric detection alone cannot see it. By the time the new sequence passes the
replica's high-water mark it looks like ordinary progress -- which is the case
the epoch exists for, and the high-water check is what catches the OTHER case
(a publisher that never learned the epoch), so both are kept.

So the identity travels WITH each message, as an opaque token in a
"<epoch>|<id>|<json>" prefix. A prefix rather than an envelope field: an older
instance would unmarshal an envelope object SILENTLY -- no matching keys, no
error, a zero-valued Event delivered to its clients -- and fails loudly on the
prefix instead.

TWO PHASES, because the failure is asymmetric. Every instance ACCEPTS both
wire forms from this release; only emission is gated, on
PAD_EVENTS_PUBLISH_EPOCH. Phase 1 rolls the binary everywhere publishing the
historical bare JSON; phase 2 sets the flag and rolls again. Flipping before
every instance is upgraded is the one direction that LOSES events rather than
resyncing: a pre-phase-1 binary cannot parse the prefix at all. Rollback is
symmetric and safe. docs/deployment.md carries the procedure both ways, what
the reset counters should read during each roll, and what remains unfixed.

Phase 2 also moves ID assignment into one atomic script. The two-call
INCR-then-PUBLISH lets two instances interleave, so a receiving instance can
append 6 before 5 -- a window older than this change, and already wrong, but
load-bearing here because counter-backwards detection reads a descending ID as
a reset. The script carries a dedupe token for the same reason
internal/watchevents' does: go-redis retries a command whose REPLY was lost, so
a publish can happen AND return an error, and the retry would deliver a second
copy that looks perfectly valid.

THE COUNTER-BACKWARDS FLOOR STAYS, and the earlier hope that this unit would
delete it was wrong. Its trigger is mixed-VERSION ordering -- an older binary
assigning and publishing in two calls -- not mixed-FORMAT payloads, so
publish-old-until-flip removes the format window only. It lives for as long as
a deployment can run two publisher versions at once, which is every rolling
upgrade, and the code now says so where it fires.

THE ASYMMETRY WITH MemoryBus IS DECLARED IN BOTH BUSES, in both packages: an
opaque epoch where the counter is shared, a numeric base where one process
owns it. They are not two spellings of one idea and must not be symmetrized.
A numeric base for Redis would close more -- it would refuse cross-incarnation
cursors, which the epoch cannot -- and is deferred rather than rejected: at the
flip, IDs would jump to ~1.8e18 in one step and every un-flipped publisher's
message would read as a massive backwards jump, dropping every buffer across
the whole roll. It is a candidate follow-on once the flip has soaked.

What this does NOT fix is stated in the code and the docs rather than implied:
the client cursor is still a bare integer with no epoch, so an old and a new ID
of the same value remain indistinguishable TO A RESUME even though the buffers
can no longer mix them.

The flip is read inside newObservedEventBus, which now takes the whole Config.
As a hand-picked argument at the two RunE call sites it was untested wiring:
replacing it with `false` compiled, passed the entire tree, and left the
deployment silently on phase 1 -- indistinguishable from a correct phase-1
deployment, since phase 1 is the default. Mutation-checked in both directions,
because a helper that ignores its config and hardcodes either value would pass
a one-directional test.

Also: the epoch and dedupe keys join the namespace assertions (an epoch shared
between two installations is a cross-feed with teeth -- each would read the
other's ID-space changes as its own), and this package's four-key EVAL is now
recorded on BUG-2724's cluster deferral, which had one call site and now has
two.

Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X

cmd/pad/cmd_server.go

@@ -687,12 +687,12 @@ func serveCmd() *cobra.Command {
 				// only trace was a line nobody's log aggregator was
 				// shaped to catch.
 				redis.SetLogger(redisSlogLogger{})
-				eventBus = newObservedEventBus(rc, redisKeys, m)
+				eventBus = newObservedEventBus(cfg, rc, redisKeys, m)
 				watchRedis = rc
 				slog.Info("Event bus using Redis pub/sub", "addr", opts.Addr, "db", opts.DB,
 					"namespace", redisKeys.Namespace())
 			} else {
-				eventBus = newObservedEventBus(nil, redisKeys, m)
+				eventBus = newObservedEventBus(cfg, nil, redisKeys, m)
 				slog.Info("Event bus using in-memory (single instance)")
 			}
 			// Wrap event bus with Prometheus instrumentation
@@ -1320,9 +1320,22 @@ func humanBytes(n int64) string {
 // restarts, and the cold-buffer resume gap is exactly as real there. Its reset
 // counter stays at zero by construction — MemoryBus owns its own IDs and has
 // no shared counter to lose.
-func newObservedEventBus(rc *redis.Client, redisKeys redisns.Keys, m *metrics.Metrics) events.EventBus {
+//
+// IT TAKES THE WHOLE CONFIG rather than the one field it needs, and that is
+// the same CONVE-19 argument one level along (BUG-2736). The phase-2 flip
+// began as a hand-picked `cfg.EventsPublishEpoch` argument at the two call
+// sites in RunE; replacing it with `false` there compiled, passed every test
+// in the tree, and left the deployment silently stuck on phase 1 —
+// indistinguishable from a correct phase-1 deployment, since phase 1 is the
+// default. Reading the field HERE puts the link inside a function a test can
+// call, and a caller that drops the config does not compile.
+//
+// The flip reaches the Redis bus only. The in-process bus has no wire and
+// identifies its ID space by its own incarnation base instead (see
+// internal/idspace), so it ignores the field.
+func newObservedEventBus(cfg *config.Config, rc *redis.Client, redisKeys redisns.Keys, m *metrics.Metrics) events.EventBus {
 	if rc != nil {
-		bus := events.NewRedisBusWithKeys(rc, redisKeys)
+		bus := events.NewRedisBusWithKeys(rc, redisKeys, cfg.EventsPublishEpoch)
 		bus.SetObserver(metrics.NewEventsObserver(m))
 		return bus
 	}

cmd/pad/event_bus_wiring_test.go

@@ -1,11 +1,15 @@
 package main
 
 import (
+	"context"
+	"encoding/json"
 	"testing"
+	"time"
 
 	"github.com/alicebob/miniredis/v2"
 	"github.com/redis/go-redis/v9"
 
+	"github.com/PerpetualSoftware/pad/internal/config"
 	"github.com/PerpetualSoftware/pad/internal/events"
 	"github.com/PerpetualSoftware/pad/internal/metrics"
 	"github.com/PerpetualSoftware/pad/internal/redisns"
@@ -22,7 +26,7 @@ import (
 func TestBothEventBusShapesReportToMetrics(t *testing.T) {
 	t.Run("in-process", func(t *testing.T) {
 		m := metrics.New()
-		bus := newObservedEventBus(nil, redisns.Default, m)
+		bus := newObservedEventBus(&config.Config{}, nil, redisns.Default, m)
 		t.Cleanup(bus.Close)
 
 		// Negative control first: a served resume must not move the counter,
@@ -53,7 +57,7 @@ func TestBothEventBusShapesReportToMetrics(t *testing.T) {
 		t.Cleanup(func() { _ = client.Close() })
 
 		m := metrics.New()
-		bus := newObservedEventBus(client, redisns.Default, m)
+		bus := newObservedEventBus(&config.Config{}, client, redisns.Default, m)
 		t.Cleanup(bus.Close)
 
 		assertResumeGaps(t, m, 0)
@@ -88,3 +92,75 @@ func assertResumeGaps(t *testing.T, m *metrics.Metrics, want float64) {
 		t.Fatalf("pad_event_resume_gaps_total = %v, want %v", got, want)
 	}
 }
+
+// The phase-2 flip has to REACH the bus, which is a separate claim from the
+// bus honouring it (BUG-2736). internal/events proves a bus constructed with
+// publishEpoch=true emits the prefixed form; this proves newObservedEventBus
+// carries the parameter there rather than dropping it, and that the
+// in-process shape ignores it instead of panicking or changing behaviour.
+//
+// The config-to-bus link is covered by the same two cases, because
+// newObservedEventBus takes the whole Config and reads the flip itself. That
+// link WAS untested when the flip was a hand-picked argument at the RunE call
+// site: replacing it with `false` there compiled and passed the whole tree,
+// leaving the deployment silently on phase 1. Both directions are asserted
+// because a helper that ignores its config and hardcodes EITHER value would
+// pass a one-directional test.
+//
+// WHAT REMAINS UNEXECUTED, stated rather than left to be assumed: the
+// `newObservedEventBus(cfg, ...)` calls inside RunE itself, which this package
+// cannot invoke without standing up a server. A separate guard
+// (redis_keyspace_wiring_test.go) reads those call sites as text and counts
+// them, which is what catches a shape being dropped.
+func TestThePublishEpochFlipReachesTheRedisBus(t *testing.T) {
+	channel := redisns.Default.Name("events:") + "ws-1"
+
+	for _, tc := range []struct {
+		name         string
+		publishEpoch bool
+		wantPrefix   bool
+	}{
+		{name: "phase 1", publishEpoch: false, wantPrefix: false},
+		{name: "phase 2", publishEpoch: true, wantPrefix: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			mr := miniredis.RunT(t)
+			client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+			t.Cleanup(func() { _ = client.Close() })
+
+			ps := client.Subscribe(context.Background(), channel)
+			t.Cleanup(func() { _ = ps.Close() })
+			if _, err := ps.Receive(context.Background()); err != nil {
+				t.Fatalf("subscribe: %v", err)
+			}
+			incoming := ps.Channel()
+
+			bus := newObservedEventBus(&config.Config{EventsPublishEpoch: tc.publishEpoch}, client, redisns.Default, metrics.New())
+			t.Cleanup(bus.Close)
+			bus.Publish(events.Event{Type: events.ItemCreated, WorkspaceID: "ws-1"})
+
+			select {
+			case msg := <-incoming:
+				// The prefixed form is not valid JSON on its own; the bare
+				// form is. That is the difference an older instance sees, so
+				// it is the difference this asserts.
+				var ev events.Event
+				bare := json.Unmarshal([]byte(msg.Payload), &ev) == nil
+				if bare == tc.wantPrefix {
+					t.Fatalf("publishEpoch=%v: got payload %q, want prefixed=%v", tc.publishEpoch, msg.Payload, tc.wantPrefix)
+				}
+			case <-time.After(2 * time.Second):
+				t.Fatal("timed out waiting for the published event")
+			}
+		})
+	}
+
+	t.Run("in-process shape ignores it", func(t *testing.T) {
+		bus := newObservedEventBus(&config.Config{EventsPublishEpoch: true}, nil, redisns.Default, metrics.New())
+		t.Cleanup(bus.Close)
+		bus.Publish(events.Event{Type: events.ItemCreated, WorkspaceID: "ws-1"})
+		if got := bus.EventsSince("ws-1", 0); len(got) != 1 {
+			t.Fatalf("the in-process bus must publish normally regardless of the flip, got %d events", len(got))
+		}
+	})
+}

docs/deployment.md

@@ -87,6 +87,7 @@ All configuration is via environment variables or a config file (`~/.pad/config.
 | `PAD_SSE_MAX_CONNECTIONS` | `1000` | Maximum streaming connections **per instance**, across both `/api/v1/events` and `/api/v1/events/stream` |
 | `PAD_SSE_MAX_PER_WORKSPACE` | `100` | Per-workspace maximum connections on `/api/v1/events`, **per instance** |
 | `PAD_SSE_MAX_PER_USER` | `50` | Per-user maximum streaming connections across both endpoints, **per instance** |
+| `PAD_EVENTS_PUBLISH_EPOCH` | `false` | Phase 2 of the event ID-space migration: publish the `<epoch>\|<id>\|<json>` wire form. **Only set this once every instance runs a binary that accepts it** — see *Event ID-space migration* below. Ignored without Redis. |
 
 #### Streaming connection limits
 
@@ -315,6 +316,84 @@ delivered. Re-sending one is a second instruction the receiving agent will act
 on twice. Only re-send a push the server told you it skipped. There is no Redis or database migration; the
 registry's keys are transient and expire on their own TTL.
 
+#### Event ID-space migration (`PAD_EVENTS_PUBLISH_EPOCH`)
+
+Events on the workspace activity stream (`GET /api/v1/events`) carry a
+`Last-Event-ID` so a reconnecting client can be replayed what it missed. With
+Redis, every instance shares one counter, so those IDs are meaningful across
+replicas.
+
+**The problem this migration fixes.** If that shared counter is ever reset —
+the key evicted under `maxmemory`, deleted by hand, a fresh Redis after a
+restore — IDs start again from 1. A replica that was buffering the old
+sequence cannot tell the new 101 from the old 101, so it can merge two ID
+spaces into one replay buffer and answer a resume across the boundary as
+though nothing was missed. Numeric detection alone cannot see it: by the time
+the new sequence passes the replica's high-water mark, it looks like ordinary
+progress.
+
+The fix gives each ID space an **epoch** — an opaque token minted with the
+space and carried on every published message, as a `<epoch>|<id>|<json>`
+prefix. A replica that sees the epoch change drops its replay buffers and
+answers resumes across the change with `sync_required`, which is honest rather
+than silent.
+
+**It rolls out in two phases, and the order is not optional.**
+
+| Phase | What you do | What instances publish | What they accept |
+|-------|-------------|------------------------|------------------|
+| 1 | Roll the new binary everywhere. Leave `PAD_EVENTS_PUBLISH_EPOCH` unset. | The historical bare JSON | Both forms |
+| 2 | Set `PAD_EVENTS_PUBLISH_EPOCH=true` and roll again. | `<epoch>\|<id>\|<json>` | Both forms |
+
+The asymmetry that makes two phases necessary: an instance running a
+**pre-phase-1** binary cannot parse a prefixed payload at all. It fails to
+unmarshal the message and drops the event for its own clients. So flipping
+before every instance is upgraded loses events on the ones that are not — not
+a resync, a silent loss.
+
+Both rolls are zero-loss in the other direction, because accept-both is on
+from phase 1: during the phase-2 roll, flipped and un-flipped instances are
+publishing different forms at the same time and every instance reads both.
+
+**Rolling back** is symmetric: unset the variable and roll. Peers accept the
+bare form throughout, so there is no window where a rollback loses events.
+There is no Redis or database migration in either direction; the epoch key is
+created by the first flipped publisher and is transient state.
+
+**What you should see when phase 2 lands.** The first flipped message reaches
+each replica and, if that replica had already buffered un-prefixed events, it
+drops its buffers once and records
+`pad_event_sequence_resets_total{reason="epoch_change"}`. Clients resuming
+across that moment get `sync_required` and re-fetch. **One drop per replica
+per roll** — if the counter keeps climbing, something is deleting the epoch or
+sequence key repeatedly; check `maxmemory-policy` against the events keyspace
+(see *Redis configuration notes*).
+
+`pad_event_sequence_resets_total{reason="counter_backward"}` is the other
+counter to watch. It fires when an ID arrives at or below what a buffer had
+already seen, which is expected in small numbers during **any** mixed-version
+roll — an older binary assigns and publishes an ID in two separate calls, so
+two instances can interleave and deliver them out of order. Phase 2 moves ID
+assignment into a single atomic Redis script, which removes the interleave for
+flipped publishers. It does **not** remove it for the roll itself: as long as
+a deployment can run two publisher versions at once, out-of-order delivery is
+possible, so this counter is expected to be non-zero during upgrades and near
+zero between them.
+
+**What this migration does not fix.** A client's `Last-Event-ID` is still a
+bare integer with no epoch in it, and that is deliberate — every deployed
+browser speaks that format, and `EventSource` echoes the header with no
+application code in the path to translate it. So an old ID and a new ID of the
+same numeric value remain indistinguishable **to a resume**, even though the
+replica's buffers can no longer mix them. The exposure is a client that
+reconnects with a cursor whose number the new sequence has already reached.
+Tracked on BUG-2736.
+
+Single-process deployments (no `PAD_REDIS_URL`) need none of this and ignore
+the variable: that bus owns its counter, so it identifies its own ID space
+from its start time and a restart's IDs cannot collide with the previous
+run's.
+
 ### Security
 
 | Variable | Default | Description |

internal/config/config.go

@@ -111,6 +111,32 @@ type Config struct {
 	// directions; see docs/deployment.md.
 	RedisNamespace string `toml:"redis_namespace"`
 
+	// EventsPublishEpoch turns on PHASE 2 of the event ID-space migration
+	// (BUG-2736): this instance publishes the "<epoch>|<id>|<json>" wire form
+	// instead of the historical bare JSON body.
+	//
+	// IT IS A TWO-PHASE FLIP, AND THE ORDER IS NOT OPTIONAL. Every instance
+	// ACCEPTS both forms from the release that introduced this field; only
+	// emission is gated. An instance running an OLDER binary cannot parse a
+	// prefixed payload at all — it fails to unmarshal and drops the event for
+	// its own clients — so flipping this before every instance is upgraded
+	// loses events for the ones that are not. Phase 1: roll the new binary
+	// everywhere with this false. Phase 2: set it true and roll again. Both
+	// forms are in flight during that second roll, which is exactly the case
+	// accept-both exists for, so it is zero-loss.
+	//
+	// What phase 2 buys, and why the emission is worth a migration: the epoch
+	// identifies WHICH INCARNATION of the shared Redis counter an event came
+	// from, so a receiving instance can tell a counter reset from ordinary
+	// progress instead of merging two ID spaces into one replay buffer. Phase
+	// 2 also moves ID assignment into a single atomic Redis script, so publish
+	// order equals ID order globally.
+	//
+	// Rolling BACK is safe in the same way and for the same reason: set it
+	// false and roll: peers accept the bare form throughout. See
+	// docs/deployment.md for the full procedure in both directions.
+	EventsPublishEpoch bool `toml:"events_publish_epoch"`
+
 	// Push carries per-USER push/consent preferences (PLAN-2613 S2). A
 	// pointer so an absent `[push]` table stays nil and Save() (via the
 	// omitempty tag) never writes an empty table into everyone's
@@ -360,6 +386,11 @@ func Load() (*Config, error) {
 	if v := os.Getenv("PAD_REDIS_NAMESPACE"); v != "" {
 		cfg.RedisNamespace = v
 	}
+	if v := os.Getenv("PAD_EVENTS_PUBLISH_EPOCH"); v != "" {
+		if on, err := strconv.ParseBool(v); err == nil {
+			cfg.EventsPublishEpoch = on
+		}
+	}
 	if v := os.Getenv("PAD_SSE_MAX_PER_USER"); v != "" {
 		if max, err := strconv.Atoi(v); err == nil {
 			cfg.SSEMaxPerUser = max

internal/config/redis_stream_config_test.go

@@ -26,6 +26,7 @@ func TestStreamAndRedisEnvMapping(t *testing.T) {
 	t.Setenv("PAD_SSE_MAX_PER_USER", "7")
 	t.Setenv("PAD_SSE_MAX_CONNECTIONS", "11")
 	t.Setenv("PAD_SSE_MAX_PER_WORKSPACE", "13")
+	t.Setenv("PAD_EVENTS_PUBLISH_EPOCH", "true")
 
 	cfg, err := Load()
 	if err != nil {
@@ -46,6 +47,30 @@ func TestStreamAndRedisEnvMapping(t *testing.T) {
 	if cfg.SSEMaxPerWorkspace != 13 {
 		t.Errorf("SSEMaxPerWorkspace = %d, want 13", cfg.SSEMaxPerWorkspace)
 	}
+	// BUG-2736's phase-2 flip. Its consumer (the Redis bus wire form) has its
+	// own tests and they all pass with Load() never populating this — the
+	// deployment would simply stay on phase 1 forever, which looks exactly
+	// like a correct phase-1 deployment. That is the wiring gap this closes.
+	if !cfg.EventsPublishEpoch {
+		t.Error("EventsPublishEpoch = false, want true from PAD_EVENTS_PUBLISH_EPOCH")
+	}
+}
+
+// A value that is not a boolean must leave the field alone rather than be
+// read as truthy. Getting this backwards flips a deployment into phase 2 on a
+// typo, which is the one direction of this migration that loses events on
+// instances that have not been upgraded.
+func TestEventsPublishEpochIgnoresANonBooleanValue(t *testing.T) {
+	t.Setenv("HOME", t.TempDir())
+	t.Setenv("PAD_EVENTS_PUBLISH_EPOCH", "yes-please")
+
+	cfg, err := Load()
+	if err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+	if cfg.EventsPublishEpoch {
+		t.Error("an unparseable value must leave the flip off, not turn it on")
+	}
 }
 
 // TestStreamAndRedisDefaults pins the shipped defaults. The per-user
@@ -55,6 +80,7 @@ func TestStreamAndRedisDefaults(t *testing.T) {
 	for _, key := range []string{
 		"PAD_REDIS_NAMESPACE", "PAD_SSE_MAX_PER_USER",
 		"PAD_SSE_MAX_CONNECTIONS", "PAD_SSE_MAX_PER_WORKSPACE",
+		"PAD_EVENTS_PUBLISH_EPOCH",
 	} {
 		if _, set := os.LookupEnv(key); set {
 			t.Setenv(key, "")
@@ -66,6 +92,13 @@ func TestStreamAndRedisDefaults(t *testing.T) {
 	if cfg.RedisNamespace != "" {
 		t.Errorf("default RedisNamespace = %q, want empty — a default namespace would move every existing deployment's keys", cfg.RedisNamespace)
 	}
+	// The default MUST be off. Phase 2 emits a wire form older instances
+	// cannot parse, so defaulting it on would break a rolling upgrade for
+	// every deployment that upgrades without reading the release notes —
+	// which is the failure the two-phase rollout exists to prevent.
+	if cfg.EventsPublishEpoch {
+		t.Error("default EventsPublishEpoch = true, want false — phase 2 must be opted into after every instance accepts the new form")
+	}
 	if cfg.SSEMaxPerUser != 50 {
 		t.Errorf("default SSEMaxPerUser = %d, want 50", cfg.SSEMaxPerUser)
 	}

internal/events/bus.go

@@ -239,6 +239,38 @@ type replayBuffer struct {
 	// ruled out on load. Do not try to recover it with a cleverer local
 	// check; there isn't one.
 	knownFrom int64
+
+	// lastAppendedID is this buffer's high-water mark. IDs reaching a given
+	// workspace's buffer ascend once every publisher is on the atomic script
+	// — the shared counter only climbs, and the script makes publish order
+	// equal ID order — but NOT during a mixed-version or mixed-FORMAT rollout,
+	// where an older build assigns and publishes as two separate calls and can
+	// deliver a LOWER ID after a higher one.
+	//
+	// So an arriving ID at or below this one means the sequence went
+	// BACKWARDS: the counter was reset, or an out-of-order delivery landed.
+	// RedisBus acts on either; MemoryBus assigns its own IDs and can never see
+	// one, which is why this field is written there and never read.
+	lastAppendedID int64
+
+	// minKnownFrom is the lowest coverage start this buffer is ALLOWED to
+	// claim, for a buffer replacing one whose sequence was discarded. Zero on
+	// an ordinary buffer. See append and newReplayBufferAfterReset.
+	minKnownFrom int64
+}
+
+// newReplayBufferAfterReset builds a buffer replacing one whose sequence was
+// discarded. It refuses every cursor BELOW discarded+1 — that is, everything
+// at or below the highest ID the discarded buffers held, EXCEPT that a cursor
+// exactly equal to `discarded` is still served, since nothing above it was
+// buffered for such a client to be missing. Pass 0 when nothing was held.
+//
+// Use newReplayBuffer for a genuinely new buffer; the two differ in what they
+// may vouch for, and that difference is the whole point of having two.
+func newReplayBufferAfterReset(size int, discarded int64) *replayBuffer {
+	rb := newReplayBuffer(size)
+	rb.minKnownFrom = discarded + 1
+	return rb
 }
 
 func newReplayBuffer(size int) *replayBuffer {
@@ -259,7 +291,34 @@ func (rb *replayBuffer) append(e Event) {
 		// First append since this buffer started (or restarted) covering
 		// the workspace. From this ID forward we can answer honestly.
 		rb.knownFrom = e.ID
+
+		// ...EXCEPT on a buffer that REPLACED one whose ID space died, where
+		// two separate assumptions behind the ordinary rule fail. Both are
+		// corrected here because both produce the same silent skip.
+		//
+		// FIRST: since() serves sinceID+1 == knownFrom on the reasoning that
+		// no ID lies strictly between the cursor and our first event — true
+		// only when both are in the SAME ID space. Across a reset they are
+		// not, so a client holding OLD 149 would be handed NEW 150 as though
+		// it followed. Hence e.ID+1: the adjacent cursor is not adjacent here.
+		//
+		// SECOND: the reset DISCARDED buffered events, and a cursor at or
+		// below the highest of them must not be told it is current — its
+		// successor is exactly what we threw away. Hence the floor.
+		//
+		// The higher of the two wins, because each is a lower bound on what
+		// this buffer can honestly claim and neither subsumes the other: the
+		// floor is based on what we HELD, e.ID+1 on what we now SEE, and a
+		// reset can move either one further out.
+		if rb.minKnownFrom > 0 {
+			rb.knownFrom = e.ID + 1
+			if rb.minKnownFrom > rb.knownFrom {
+				rb.knownFrom = rb.minKnownFrom
+			}
+			rb.minKnownFrom = 0
+		}
 	}
+	rb.lastAppendedID = e.ID
 }
 
 // since returns all buffered events with ID > sinceID, in chronological order.
@@ -354,6 +413,23 @@ type MemoryBus struct {
 	// construction; see internal/idspace for what it buys and what it costs.
 	// A zero base is the pre-BUG-2736 behaviour (IDs from 1) and is
 	// deliberately unreachable through any constructor.
+	//
+	// WHY THIS BUS USES A NUMERIC BASE AND RedisBus USES AN OPAQUE EPOCH.
+	// They are not two spellings of one idea and must not be symmetrized into
+	// one (BUG-2736). This bus is the SOLE publisher into its space, so it can
+	// compute an identity at startup and put it in the ID's value, which makes
+	// a cross-incarnation cursor numerically refusable for free. RedisBus's
+	// counter is shared across processes: no instance can compute an identity
+	// the others would agree with, so the identity travels WITH each message.
+	//
+	// A numeric base for the Redis counter would close more (it would refuse
+	// cross-incarnation cursors there too, which the epoch cannot), and it was
+	// weighed and deferred: at the phase-2 flip, IDs would jump from small to
+	// ~1.8e18 in one step, so every publish from an un-flipped instance would
+	// read as a massive backwards jump and drop every buffer — a resync storm
+	// spanning the whole roll. It is a candidate follow-on once the flip has
+	// shipped and soaked, when no un-flipped publisher remains to misread the
+	// jump. BUG-2736's trail carries the reasoning.
 	base int64
 
 	// Per-workspace replay buffers for Last-Event-ID support.

internal/events/observer.go

@@ -77,6 +77,28 @@ const (
 	// ResetReasonSubscriptionResumed is scoped to ONE workspace: a dropped
 	// subscription says nothing about any other channel. See Observer.
 	ResetReasonSubscriptionResumed = "subscription_resumed"
+
+	// ResetReasonEpochChange means the shared Redis counter's ID SPACE
+	// changed: the epoch travelling with an arriving message is not the one
+	// this instance had adopted (BUG-2736). Every buffer is dropped, not just
+	// the arriving event's workspace, because the counter is global.
+	//
+	// What an operator does with it: a handful at once, correlated with a
+	// deploy or a Redis restart, is the mechanism working. A steady trickle
+	// means the counter key is being evicted or deleted repeatedly — check
+	// maxmemory policy against the events keyspace.
+	ResetReasonEpochChange = "epoch_change"
+
+	// ResetReasonCounterBackward means an ID arrived at or below a buffer's
+	// high-water mark WITHOUT an epoch change: the same numeric space
+	// delivered something out of order, or restarted inside its own epoch.
+	//
+	// EXPECTED DURING A MIXED-VERSION OR MIXED-FORMAT ROLL and not otherwise:
+	// an older publisher assigns and publishes in two calls, so two instances
+	// can interleave. Seeing it in steady state, with every instance on the
+	// atomic script, is an anomaly worth investigating rather than tuning
+	// away — see the comment at the branch that reports it.
+	ResetReasonCounterBackward = "counter_backward"
 )
 
 // observable is the shared, nil-safe Observer holder both bus implementations

internal/events/redis_bus.go

@@ -3,11 +3,15 @@ package events
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"log/slog"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 
 	"github.com/PerpetualSoftware/pad/internal/redisns"
+	"github.com/google/uuid"
 	"github.com/redis/go-redis/v9"
 )
 
@@ -25,8 +29,115 @@ const (
 	// the new counter starts from zero and connected clients' Last-Event-ID
 	// values belong to the old space.
 	redisSeqSuffix = "event_seq"
+
+	// redisEpochSuffix identifies the CURRENT ID space ("pad:event_epoch" by
+	// default), and exists because numeric detection alone cannot see a reset
+	// that has already caught back up.
+	//
+	// A counter reset is detectable when an ID arrives at or below our
+	// high-water mark. It is INVISIBLE when the new space has already climbed
+	// past it: hold 100, lose the subscription, the counter resets and IDs
+	// 1-101 are published, and the only one that reaches us is 101 — which
+	// looks exactly like the contiguous successor of 100. The buffer then
+	// mixes two ID spaces and a client resuming from OLD 100 is handed NEW 101,
+	// having silently missed everything the old space had above 100.
+	//
+	// Same mechanism, same reasoning, and the same key shape as
+	// internal/watchevents' watchevents_epoch. Deliberately a DISTINCT key
+	// from that one for the same reason the counters are distinct: the two
+	// buses carry independent Last-Event-ID spaces.
+	redisEpochSuffix = "event_epoch"
+
+	// redisDedupeSuffix namespaces the per-publish idempotency tokens.
+	redisDedupeSuffix = "events:pub:"
+
+	// redisDedupeTTLSeconds bounds how long a token is remembered. It only has
+	// to cover a client-side retry burst — go-redis gives up after MaxRetries
+	// with backoff measured in milliseconds — so a minute is generous, and the
+	// keys are small and expire on their own.
+	redisDedupeTTLSeconds = 60
 )
 
+// DEPLOYMENT SCOPING (BUG-2724). Every name above carries the installation's
+// PAD_REDIS_NAMESPACE when one is set — see internal/redisns — and is
+// byte-identical to the historical flat names when it is not. The rule, stated
+// the same way in internal/watchevents and internal/server's presence registry
+// because it belongs to all three at once: scoping comes from ONE shared
+// config value built in cmd/pad/cmd_server.go, never from one package growing
+// a prefix the others lack.
+//
+// STILL UNSCOPED, deliberately: Redis CLUSTER. No hash tags, and a non-cluster
+// client. BUG-2736 adds a SECOND multi-key EVAL to the codebase — publishScript
+// here spans four keys (sequence, channel, epoch, dedupe), matching the one in
+// internal/watchevents — so a cluster port now has two call sites that would
+// fail CROSSSLOT rather than one. Same deferral, one more site; BUG-2724 holds
+// the reasoning for shipping tags only alongside a cluster client that can
+// test them.
+
+// publishScript assigns the ID and publishes in ONE atomic Redis call. It is
+// PHASE 2 ONLY: an instance that has not been flipped still publishes through
+// the two-call path in Publish, because the bare wire form carries the ID
+// INSIDE the JSON and the JSON must therefore be marshalled after the ID is
+// known. See config.EventsPublishEpoch for the rollout order.
+//
+// WHY ATOMIC. The two-call version does INCR and PUBLISH as separate
+// round-trips, which lets two instances interleave — INCR 5, INCR 6, PUBLISH
+// 6, PUBLISH 5 — so a receiving instance can append 6 before 5 and corrupt the
+// ordering replayBuffer.since() assumes when it computes oldest and newest.
+// That window is older than this fix and was already wrong; it becomes
+// load-bearing here, because counter-backwards detection reads a descending ID
+// as a RESET, and under the two-call version every interleave would look like
+// one. Redis runs a script atomically on its single thread, so publish order
+// equals ID order globally with no coordination on our side.
+//
+// THE EPOCH AND ID ARE PREPENDED as "<epoch>|<id>|<json>" rather than injected
+// into the JSON. Two reasons, and the second is the one a future refactor to
+// "cleaner JSON" would regress: string-editing JSON inside Lua is fragile, and
+// an envelope object would be UNMARSHALLED SILENTLY by an older instance
+// during a mixed roll — no matching keys, no error, a zero-valued Event
+// delivered to that instance's clients. The prefix fails loudly instead. See
+// decodePayload for the receiving side, which accepts both forms.
+//
+// The epoch is offered SET NX in steady state, so every publisher can propose
+// a candidate and exactly the first one wins. The id == 1 branch is the
+// deliberate exception and overwrites it — see that branch.
+//
+// A DEDUPE TOKEN, matching internal/watchevents' script. The mechanism it
+// closes: go-redis retries a command whose reply was lost to a network error —
+// a Redis failover being the obvious trigger — so the script can run, publish,
+// and still return an error to its caller. Retried, it would publish the same
+// event again under a second ID, and both copies look perfectly valid
+// (ascending IDs, correct ordering), so nothing downstream can tell them
+// apart. It is not merely a duplicate row: the web layout raises a toast for
+// any externally-sourced item_created, so a duplicate is a duplicate toast
+// plus a redundant fetch.
+//
+// SET NX on a caller-generated token turns the retry into a no-op: the retry
+// carries the same KEYS[4], the SET fails, and the script returns 0 without
+// publishing.
+var publishScript = redis.NewScript(`
+if redis.call('SET', KEYS[4], '1', 'NX', 'EX', ARGV[3]) == false then
+  return 0
+end
+local id = redis.call('INCR', KEYS[1])
+if id == 1 then
+  -- The counter is starting from scratch: this installation's first publish
+  -- ever, or the seq key was deleted or evicted under us. Both are a NEW id
+  -- space, so the epoch is ROTATED rather than merely offered. Without this, a
+  -- deleted seq key restarts the ids inside the SAME epoch and the epoch check
+  -- reports nothing -- and the numeric check misses it too whenever a
+  -- receiver's high-water mark is low enough that the restarted counter climbs
+  -- past it before that receiver sees anything.
+  redis.call('SET', KEYS[3], ARGV[2])
+else
+  -- Steady state: offer a candidate, and let the first publisher win.
+  redis.call('SET', KEYS[3], ARGV[2], 'NX')
+end
+local epoch = redis.call('GET', KEYS[3])
+redis.call('PUBLISH', KEYS[2], epoch .. '|' .. id .. '|' .. ARGV[1])
+return id
+`)
+
 // RedisBus distributes events across multiple Pad instances via Redis pub/sub.
 // Each instance subscribes to Redis channels for its locally-connected SSE clients,
 // and publishes events to Redis so all instances see them.
@@ -89,6 +200,41 @@ type RedisBus struct {
 	replayBuffers map[string]*replayBuffer
 	replaySize    int
 
+	// publishEpoch selects the wire form this instance EMITS: the phase-2
+	// "<epoch>|<id>|<json>" prefix when true, the historical bare JSON body
+	// when false. Receiving accepts both regardless — see decodePayload and
+	// config.EventsPublishEpoch for why emission is the half that is gated.
+	publishEpoch bool
+
+	// epoch is the ID space this instance has adopted, learned from arriving
+	// messages rather than read at startup.
+	//
+	// AN OPAQUE TOKEN, NOT A NUMERIC BASE, and the asymmetry with MemoryBus is
+	// deliberate rather than drift — see the `base` field on MemoryBus for the
+	// full reasoning. In one sentence: this counter is SHARED across
+	// processes, so no instance can compute an identity the others would
+	// agree with, and the identity has to travel with the message instead. The
+	// cost of that choice is that a cursor still carries no space of its own,
+	// so an old and a new ID of the same value remain indistinguishable to a
+	// resume; BUG-2736's trail names the numeric base that would close it and
+	// why it is a follow-on unit. Empty until a prefixed message
+	// arrives, which on a phase-1 deployment is never. Guarded by mu.
+	//
+	// LEARNED, NOT FETCHED, deliberately: reading the key at construction
+	// would make an instance believe it belongs to a space whose events it has
+	// not received, which is precisely the claim BUG-2731 spent a unit
+	// removing. The epoch matters only in relation to buffered events, so it
+	// arrives with them.
+	epoch string
+
+	// hadReset and discardedHighWater record that this bus has thrown a
+	// sequence away, and how high the discarded buffers had climbed. Every
+	// buffer built afterwards refuses cursors at or below that mark — see
+	// newBuffer and dropAllBuffers, where the two reset reasons set them
+	// differently on purpose. Guarded by mu.
+	hadReset           bool
+	discardedHighWater int64
+
 	ctx    context.Context
 	cancel context.CancelFunc
 }
@@ -106,18 +252,25 @@ type redisSub struct {
 // NewRedisBus creates a new Redis-backed EventBus.
 // The provided redis.Client should already be configured and connected.
 func NewRedisBus(client *redis.Client) *RedisBus {
-	return NewRedisBusWithKeys(client, redisns.Default)
+	return NewRedisBusWithKeys(client, redisns.Default, false)
 }
 
 // NewRedisBusWithKeys is NewRedisBus with an explicit key namespace
 // (BUG-2724). cmd/pad/cmd_server.go uses this one, passing the value
 // shared with the watch bus and the presence registry so all three
 // keyspaces carry the same namespace or none.
-func NewRedisBusWithKeys(client *redis.Client, keys redisns.Keys) *RedisBus {
+//
+// publishEpoch selects the wire form this instance EMITS (BUG-2736). It is a
+// constructor parameter with no default rather than a setter, so every call
+// site states which phase of the rollout it is in and none can flip a bus that
+// is already publishing. See config.EventsPublishEpoch for the order the two
+// phases must be rolled in.
+func NewRedisBusWithKeys(client *redis.Client, keys redisns.Keys, publishEpoch bool) *RedisBus {
 	ctx, cancel := context.WithCancel(context.Background())
 	return &RedisBus{
 		client:        client,
 		keys:          keys,
+		publishEpoch:  publishEpoch,
 		subscribers:   make(map[string]map[chan Event]*subscriber),
 		workspaceOf:   make(map[chan Event]string),
 		wsCounts:      make(map[string]int),
@@ -217,9 +370,21 @@ func (b *RedisBus) Publish(event Event) {
 	}
 
 	channel := b.keys.Name(redisChannelSuffix) + event.WorkspaceID
-	// Assign a globally ordered sequence ID via Redis atomic counter, so all
-	// instances share one ID space and Last-Event-ID from one instance is
-	// meaningful on any other.
+	if b.publishEpoch {
+		b.publishWithEpoch(channel, event)
+		return
+	}
+
+	// PHASE 1: the historical two-call path, unchanged. Assign a globally
+	// ordered sequence ID via Redis atomic counter, so all instances share one
+	// ID space and Last-Event-ID from one instance is meaningful on any other.
+	//
+	// It keeps the pre-existing interleave window (INCR and PUBLISH are two
+	// round-trips, so two instances can publish out of ID order) and the
+	// pre-existing retry-duplication window. Both are closed by phase 2 rather
+	// than here, because the bare wire form carries the ID INSIDE the JSON and
+	// the JSON therefore cannot be marshalled until the ID is known — which is
+	// the whole reason the atomic script and the prefix arrive together.
 	id, err := b.client.Incr(b.ctx, b.keys.Name(redisSeqSuffix)).Result()
 	if err != nil {
 		// NO LOCAL-COUNTER FALLBACK, and its removal is a fix rather than a
@@ -246,6 +411,44 @@ func (b *RedisBus) Publish(event Event) {
 	}
 }
 
+// publishWithEpoch is the PHASE 2 path: one atomic script assigns the ID,
+// maintains the epoch, and publishes "<epoch>|<id>|<json>".
+//
+// The event is marshalled with ID still zero, because the ID travels in the
+// prefix and decodePayload writes it back onto the decoded Event. A receiver
+// running phase 1 or phase 2 reads the same value either way; a receiver
+// running a PRE-phase-1 binary cannot parse this at all, which is the reason
+// the flip is a second roll rather than a config change (see
+// config.EventsPublishEpoch).
+func (b *RedisBus) publishWithEpoch(channel string, event Event) {
+	data, err := json.Marshal(event)
+	if err != nil {
+		slog.Error("failed to marshal event for Redis", "error", err)
+		return
+	}
+
+	// A fresh token per logical publish — NOT per attempt, which is the point:
+	// go-redis reuses the same arguments on its own retries, so the second run
+	// of the script sees the same token and declines.
+	dedupeKey := b.keys.Name(redisDedupeSuffix) + uuid.NewString()
+	if err := publishScript.Run(b.ctx, b.client,
+		[]string{b.keys.Name(redisSeqSuffix), channel, b.keys.Name(redisEpochSuffix), dedupeKey},
+		string(data), uuid.NewString(), redisDedupeTTLSeconds).Err(); err != nil {
+		// NO LOCAL-COUNTER FALLBACK, for the same reason the phase-1 path has
+		// none (BUG-2731): an ID minted locally belongs to a different space,
+		// which every receiving instance reads as a counter reset, and this
+		// bus has no local fan-out path so the event reaches nobody here
+		// either way.
+		//
+		// Note what this error does and does not mean: the script is atomic,
+		// so it never half-executes — but go-redis retries a command whose
+		// REPLY was lost, so an error here can accompany a publish that
+		// actually happened. That is what the dedupe token is for, and why
+		// this logs rather than re-publishing.
+		slog.Error("failed to publish event to Redis", "channel", channel, "error", err)
+	}
+}
+
 // EventsSince returns buffered events for a workspace with IDs greater than
 // sinceID. Returns nil when this instance cannot vouch for the requested span
 // — see the EventBus interface and replayBuffer.since.
@@ -429,12 +632,12 @@ func (b *RedisBus) receiveMessages(ctx context.Context, pubsub *redis.PubSub, wo
 				b.dropWorkspaceCoverage(workspaceID, ResetReasonSubscriptionResumed, gen)
 
 			case *redis.Message:
-				var event Event
-				if err := json.Unmarshal([]byte(msg.Payload), &event); err != nil {
-					slog.Error("failed to unmarshal Redis event", "channel", msg.Channel, "error", err)
+				epoch, event, err := decodePayload(msg.Payload)
+				if err != nil {
+					slog.Error("failed to decode Redis event", "channel", msg.Channel, "error", err)
 					continue
 				}
-				b.fanOutFromRedis(gen, event)
+				b.fanOutFromRedis(gen, epoch, event)
 			}
 		}
 	}
@@ -495,16 +698,135 @@ func (b *RedisBus) currentSubGen(workspaceID string) int64 {
 	return 0
 }
 
+// buffersHoldEvents reports whether any replay buffer has been written to.
+// Callers must hold mu.
+func (b *RedisBus) buffersHoldEvents() bool {
+	for _, rb := range b.replayBuffers {
+		if rb.count > 0 {
+			return true
+		}
+	}
+	return false
+}
+
+// newBuffer builds a replay buffer of the right flavour for this bus's
+// history: once a sequence has been discarded under us, every buffer we build
+// afterwards must refuse the cursor immediately below its first event, because
+// that cursor may belong to the sequence we threw away. Callers must hold mu.
+func (b *RedisBus) newBuffer() *replayBuffer {
+	if b.hadReset {
+		return newReplayBufferAfterReset(b.replaySize, b.discardedHighWater)
+	}
+	return newReplayBuffer(b.replaySize)
+}
+
+// dropAllBuffers throws away every replay buffer because this instance can no
+// longer vouch for the sequence they describe — either the ID space itself
+// changed, or an ID arrived at or below the high-water mark. Callers must hold
+// mu.
+//
+// raiseFloor says whether the replacements should additionally refuse every
+// cursor at or below what the discarded buffers held, and the two reset
+// reasons answer it differently ON PURPOSE. Getting this backwards produces a
+// RESYNC LOOP, which is worse than the bug the floor exists to fix.
+//
+//   - COUNTER BACKWARDS, no epoch change: the arriving ID is in the SAME
+//     numeric space we were already tracking (whoever published it INCRed the
+//     same key). We just discarded events a cursor can legitimately ask about,
+//     and the successor of such a cursor is exactly what we threw away — so
+//     the floor is raised. The case this really covers is a mixed-version or
+//     mixed-FORMAT roll, where a phase-1 publisher's non-atomic INCR/PUBLISH
+//     delivers a LOWER ID after a newer one.
+//
+//   - EPOCH CHANGE: a genuinely NEW space, typically restarting from 1 while
+//     the dead space had climbed high. Raising the floor there would refuse
+//     every cursor until the new counter passed the old high-water mark — and
+//     since each refusal hands the client a FRESH low cursor that is refused
+//     again, that is a loop, not one resync. The ambiguity between an old and
+//     a new ID of the same value is accepted instead, exactly as
+//     internal/watchevents accepts it; see BUG-2736's trail for the numeric
+//     base that would close it and why it is not this unit.
+//
+// An epoch change also CLEARS any standing floor, which is not the same as
+// declining to raise one. The floor is a same-space device: it names IDs whose
+// successors we discarded FROM THE SPACE WE WERE TRACKING. Once the space
+// itself is gone those numbers mean nothing, and leaving the floor standing
+// produces the very loop the paragraph above rules out — just reached from a
+// bus that took a counter-backwards reset earlier in its life.
+func (b *RedisBus) dropAllBuffers(raiseFloor bool) {
+	if raiseFloor {
+		for _, rb := range b.replayBuffers {
+			if rb.lastAppendedID > b.discardedHighWater {
+				b.discardedHighWater = rb.lastAppendedID
+			}
+		}
+	} else {
+		b.discardedHighWater = 0
+	}
+	b.hadReset = true
+	b.replayBuffers = make(map[string]*replayBuffer)
+}
+
+// decodePayload parses the "<epoch>|<id>|<json>" wire form publishScript emits,
+// and ALSO accepts a bare JSON body with no prefix.
+//
+// THE BARE FORM IS NOT LEGACY-ONLY. It is what every phase-1 instance
+// publishes — which, until an operator flips config.EventsPublishEpoch, is
+// every instance — as well as what a pre-BUG-2736 binary publishes. Accepting
+// it is what makes both rolls zero-loss in the new-receiving-old direction. It
+// returns an empty epoch, which the receive path reads as "no ID-space
+// information" and leaves the epoch bookkeeping untouched rather than treating
+// it as a change.
+//
+// The reverse direction is not recoverable from this side: an instance running
+// a PRE-phase-1 binary fails to unmarshal a prefixed payload and drops the
+// event for its own clients, loudly. That asymmetry is the entire reason the
+// flip is a second roll. See docs/deployment.md.
+//
+// Splitting on the FIRST two separators keeps a '|' inside the JSON body
+// harmless: the epoch is a uuid and the ID is digits, so neither can contain
+// one. The leading '{' check is what stops a JSON body that happens to contain
+// two '|' characters from being mistaken for a prefixed payload — an epoch is
+// never a JSON object.
+func decodePayload(payload string) (string, Event, error) {
+	if parts := strings.SplitN(payload, "|", 3); len(parts) == 3 && !strings.HasPrefix(parts[0], "{") {
+		epoch, idPart, body := parts[0], parts[1], parts[2]
+		if epoch == "" {
+			return "", Event{}, fmt.Errorf("payload has an empty epoch prefix")
+		}
+		id, err := strconv.ParseInt(idPart, 10, 64)
+		if err != nil {
+			return "", Event{}, fmt.Errorf("payload id prefix %q is not an integer: %w", idPart, err)
+		}
+		var event Event
+		if err := json.Unmarshal([]byte(body), &event); err != nil {
+			return "", Event{}, fmt.Errorf("payload body is not an Event: %w", err)
+		}
+		event.ID = id
+		return epoch, event, nil
+	}
+
+	var event Event
+	if err := json.Unmarshal([]byte(payload), &event); err != nil {
+		return "", Event{}, fmt.Errorf("payload is neither <epoch>|<id>|<json> nor a bare Event: %w", err)
+	}
+	return "", event, nil
+}
+
 // fanOutFromRedis is the receive path: a message that arrived on the
-// subscription identified by gen.
-func (b *RedisBus) fanOutFromRedis(gen int64, event Event) {
-	b.fanOut(gen, event)
+// subscription identified by gen, carrying the ID space it belongs to.
+//
+// An empty epoch means the payload carried no ID-space information (a phase-1
+// or pre-BUG-2736 publisher, or a direct test call) and the bookkeeping is left
+// alone — silence is not evidence of a change.
+func (b *RedisBus) fanOutFromRedis(gen int64, epoch string, event Event) {
+	b.fanOut(gen, epoch, event)
 }
 
 // fanOutLocally distributes an event to all local subscribers for the event's
 // workspace and stores it in the replay buffer, with no id-space information.
 func (b *RedisBus) fanOutLocally(event Event) {
-	b.fanOut(anySubscription, event)
+	b.fanOut(anySubscription, "", event)
 }
 
 // anySubscription opts out of the generation check for callers that are not a
@@ -512,7 +834,7 @@ func (b *RedisBus) fanOutLocally(event Event) {
 // carries the generation of the subscription it arrived on.
 const anySubscription int64 = 0
 
-func (b *RedisBus) fanOut(gen int64, event Event) {
+func (b *RedisBus) fanOut(gen int64, epoch string, event Event) {
 	// Registered FIRST so it runs LAST — after the Unlock below, so an
 	// observer may call back into the bus without deadlocking the receive
 	// loop.
@@ -555,12 +877,102 @@ func (b *RedisBus) fanOut(gen int64, event Event) {
 		return
 	}
 
+	// ID-SPACE RECONCILIATION (BUG-2736). Both checks answer the same question
+	// — do the events already buffered belong to the same sequence as this
+	// one? — and both are needed, because neither sees the other's case. The
+	// epoch catches a reset that has already climbed past our high-water mark,
+	// which is numerically invisible; the high-water check catches a reset on
+	// an instance that never learned the previous epoch, including one
+	// publishing from a phase-1 or pre-BUG-2736 binary.
+	//
+	// Every buffer is dropped, not just this workspace's: the counter is
+	// global, so a reset invalidates all of them at once.
+	if epoch != "" && b.epoch != epoch {
+		// ADOPTING AN EPOCH ONTO A NON-EMPTY BUFFER IS ALSO A RESET. Learning
+		// an epoch for the first time normally means the first message of this
+		// bus's life, and dropping empty buffers would be pointless. But
+		// during the phase-2 roll the buffers can already hold events from a
+		// phase-1 publisher, whose payloads carry no epoch at all — and those
+		// events' ID space is exactly what we have no way to compare against
+		// the one we are now being told about.
+		//
+		// The dangerous shape: bare events up to 5, the counter is then
+		// deleted, a flipped publisher rotates the epoch and climbs to 6
+		// before this instance receives anything. 6 exceeds our high-water
+		// mark, so the numeric check sees an ordinary successor, and without
+		// this branch the two spaces merge in one buffer.
+		//
+		// Costs at most ONE drop per instance per roll: once adopted, later
+		// bare messages leave the epoch alone and later prefixed ones match.
+		//
+		// WHAT THIS DELIBERATELY DOES NOT COVER: a bus whose buffers are
+		// EMPTY adopts without dropping, so its first buffer starts at exactly
+		// the first ID it sees. A client holding the ID one below that — from
+		// a space this process never saw, because it started through a
+		// cutover — is then served. Closing it locally means every bus
+		// refusing the adjacent cursor forever, which trades a RARE silent
+		// skip for a COMMON extra resync: on a multi-instance deployment a
+		// client legitimately holds ID 149 from replica A and reconnects to
+		// replica B whose first ID for that workspace is 150, and consecutive
+		// global IDs land in the same busy workspace routinely. BUG-2736's
+		// trail names the numeric-base design that closes it with neither
+		// cost, and why it is a follow-on rather than this unit.
+		if b.epoch != "" || b.buffersHoldEvents() {
+			slog.Warn("event ID space changed; dropping replay buffers, resumes spanning the change will report sync_required",
+				"previous_epoch", b.epoch, "new_epoch", epoch, "id", event.ID)
+			b.dropAllBuffers(false)
+			reset = ResetReasonEpochChange
+		}
+		b.epoch = epoch
+	}
+
 	// Store in replay buffer for reconnect replay.
 	rb, ok := b.replayBuffers[event.WorkspaceID]
 	if !ok {
-		rb = newReplayBuffer(b.replaySize)
+		rb = b.newBuffer()
 		b.replayBuffers[event.WorkspaceID] = rb
 	}
+	if rb.lastAppendedID != 0 && event.ID <= rb.lastAppendedID {
+		// THIS WHOLE MECHANISM IS TRANSITIONAL. READ THIS BEFORE TUNING IT.
+		//
+		// Once every publisher is flipped to phase 2, publish order equals ID
+		// order globally, and a genuine counter restart rotates the epoch (see
+		// publishScript's id == 1 branch), which is a different code path
+		// entirely. So a backwards ID in steady state is an ANOMALY, not a
+		// case needing clever classification. Its real job is the ROLL WINDOW,
+		// where a phase-1 publisher's non-atomic INCR/PUBLISH can deliver a
+		// lower ID after a higher one.
+		//
+		// IT DOES NOT GO AWAY WHEN THE FORMAT FLIP COMPLETES, which the
+		// original scoping hoped it would. The trigger is mixed-VERSION
+		// ORDERING — older binaries assign and publish in two calls — not
+		// mixed-format payloads, so publish-old-until-flip removes the
+		// mixed-FORMAT window only. The floor lives for as long as any
+		// deployment can run two publisher versions at once, which is every
+		// rolling upgrade.
+		//
+		// It has already been tuned four times, each round finding a defect
+		// inside the previous round's cleverness. So the floor is raised
+		// UNCONDITIONALLY: every cursor at or below what the discarded buffers
+		// held is refused. The alternative was a discriminator on "a restart
+		// always begins at 1", which an out-of-order ID 1 during a roll
+		// defeats — silently. Refusing too much is loud: it shows up in
+		// pad_event_resume_gaps_total and in the warning below. Refusing too
+		// little loses events with nothing to show for it.
+		//
+		// THE COST, stated so nobody rediscovers it as a bug: if a phase-1
+		// publisher restarts the counter mid-roll, cursors are refused until
+		// the sequence climbs past the dead high-water mark, so affected
+		// clients resync repeatedly. Bounded by the roll — once every
+		// publisher is flipped, restarts rotate the epoch, and an epoch change
+		// CLEARS the floor (see dropAllBuffers).
+		slog.Warn("event sequence went backwards; dropping replay buffers, resumes below the discarded high-water mark will report sync_required",
+			"high_water_mark", rb.lastAppendedID, "id", event.ID, "workspace", event.WorkspaceID)
+		b.dropAllBuffers(true)
+		rb = b.newBuffer()
+		b.replayBuffers[event.WorkspaceID] = rb
+		reset = ResetReasonCounterBackward
+	}
 	rb.append(event)
 
 	for _, sub := range b.subscribers[event.WorkspaceID] {

internal/events/redis_idspace_test.go

@@ -0,0 +1,387 @@
+package events
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/PerpetualSoftware/pad/internal/redisns"
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+)
+
+// BUG-2736's Redis half. The counter is shared ACROSS processes, so no single
+// instance can compute which incarnation of it an ID belongs to — the way
+// MemoryBus does with its own base. The identity therefore travels WITH each
+// message, as an opaque epoch in a "<epoch>|<id>|<json>" prefix, and emitting
+// it is gated behind a two-phase flip because an older binary cannot parse it.
+
+// listen subscribes to a channel with a REAL go-redis client and returns a
+// function that reads the next published payload. Using the real client rather
+// than poking miniredis keeps the bytes under test the bytes a receiving
+// instance would actually see.
+func listen(t *testing.T, client *redis.Client, channel string) func() string {
+	t.Helper()
+	ps := client.Subscribe(context.Background(), channel)
+	t.Cleanup(func() { _ = ps.Close() })
+	if _, err := ps.Receive(context.Background()); err != nil {
+		t.Fatalf("subscribe to %s: %v", channel, err)
+	}
+	ch := ps.Channel()
+	return func() string {
+		select {
+		case msg, ok := <-ch:
+			if !ok {
+				t.Fatal("subscription closed before a message arrived")
+			}
+			return msg.Payload
+		case <-time.After(2 * time.Second):
+			t.Fatal("timed out waiting for a published message")
+			return ""
+		}
+	}
+}
+
+func newFlippedRedisBus(t *testing.T) (*RedisBus, *miniredis.Miniredis) {
+	t.Helper()
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	t.Cleanup(func() { _ = client.Close() })
+	b := NewRedisBusWithKeys(client, redisns.Default, true)
+	t.Cleanup(b.Close)
+	return b, mr
+}
+
+// --- the wire form ------------------------------------------------------
+
+func TestDecodePayloadAcceptsBothWireForms(t *testing.T) {
+	// Both directions of the roll depend on this function, so every branch is
+	// pinned rather than the happy one.
+	bare, err := json.Marshal(Event{ID: 42, Type: ItemUpdated, WorkspaceID: "ws-1"})
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+
+	t.Run("prefixed", func(t *testing.T) {
+		body, _ := json.Marshal(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
+		epoch, ev, err := decodePayload("e-1|77|" + string(body))
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if epoch != "e-1" {
+			t.Fatalf("epoch: want e-1, got %q", epoch)
+		}
+		// The ID comes from the PREFIX, not the body — the body was
+		// marshalled before the ID existed.
+		if ev.ID != 77 {
+			t.Fatalf("id: want 77, got %d", ev.ID)
+		}
+		if ev.WorkspaceID != "ws-1" {
+			t.Fatalf("workspace: want ws-1, got %q", ev.WorkspaceID)
+		}
+	})
+
+	t.Run("bare, which is what every phase-1 publisher emits", func(t *testing.T) {
+		epoch, ev, err := decodePayload(string(bare))
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if epoch != "" {
+			t.Fatalf("a bare payload carries no id-space information; got epoch %q", epoch)
+		}
+		if ev.ID != 42 {
+			t.Fatalf("id: want 42 from the body, got %d", ev.ID)
+		}
+	})
+
+	t.Run("a bare body containing pipes is not mistaken for a prefix", func(t *testing.T) {
+		// The discriminating case for the leading-'{' check. Without it this
+		// JSON splits into three parts and is rejected as a bad prefix.
+		body, _ := json.Marshal(Event{ID: 9, Type: ItemUpdated, WorkspaceID: "ws-1", Title: "a|b|c"})
+		if !strings.Contains(string(body), "|") {
+			t.Fatal("fixture: the body must contain pipes for this case to mean anything")
+		}
+		epoch, ev, err := decodePayload(string(body))
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if epoch != "" || ev.ID != 9 || ev.Title != "a|b|c" {
+			t.Fatalf("want bare decode with title intact, got epoch=%q id=%d title=%q", epoch, ev.ID, ev.Title)
+		}
+	})
+
+	t.Run("rejections", func(t *testing.T) {
+		body, _ := json.Marshal(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
+		for name, payload := range map[string]string{
+			"empty epoch":       "|77|" + string(body),
+			"non-integer id":    "e-1|seventy|" + string(body),
+			"body is not JSON":  "e-1|77|not json",
+			"neither form":      "not json at all",
+			"prefix without id": "e-1|" + string(body),
+		} {
+			if _, _, err := decodePayload(payload); err == nil {
+				t.Errorf("%s: want an error, got none", name)
+			}
+		}
+	})
+}
+
+func TestPhaseTwoPublishesThePrefixedFormAndPhaseOneDoesNot(t *testing.T) {
+	// The flip's ONLY observable difference is the bytes on the wire, so this
+	// reads them rather than any internal flag.
+	t.Run("flipped", func(t *testing.T) {
+		b, _ := newFlippedRedisBus(t)
+		next := listen(t, b.client, redisns.Default.Name(redisChannelSuffix)+"ws-1")
+
+		b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
+
+		epoch, ev, err := decodePayload(next())
+		if err != nil {
+			t.Fatalf("a flipped instance must emit a decodable payload: %v", err)
+		}
+		if epoch == "" {
+			t.Fatal("a flipped instance must emit an epoch")
+		}
+		if ev.ID != 1 {
+			t.Fatalf("first id from a fresh counter must be 1, got %d", ev.ID)
+		}
+	})
+
+	t.Run("not flipped", func(t *testing.T) {
+		mr := miniredis.RunT(t)
+		client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+		t.Cleanup(func() { _ = client.Close() })
+		b := NewRedisBusWithKeys(client, redisns.Default, false)
+		t.Cleanup(b.Close)
+
+		next := listen(t, client, redisns.Default.Name(redisChannelSuffix)+"ws-1")
+
+		b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
+
+		// A pre-phase-1 binary must be able to json.Unmarshal this directly —
+		// that is the entire compatibility claim, so assert it the way that
+		// binary would rather than through decodePayload, which accepts both.
+		var ev Event
+		if err := json.Unmarshal([]byte(next()), &ev); err != nil {
+			t.Fatalf("an un-flipped instance must emit bare JSON an older binary can parse: %v", err)
+		}
+		if ev.ID != 1 {
+			t.Fatalf("the bare form carries the id INSIDE the body; want 1, got %d", ev.ID)
+		}
+		if mr.Exists(redisns.Default.Name(redisEpochSuffix)) {
+			t.Fatal("an un-flipped instance must not write the epoch key")
+		}
+	})
+}
+
+func TestTheDedupeTokenMakesARetriedPublishANoOp(t *testing.T) {
+	// go-redis retries a command whose REPLY was lost, so the script can run,
+	// publish, and still return an error to its caller. Driving the script
+	// twice with the same token is that retry.
+	b, mr := newFlippedRedisBus(t)
+	channel := redisns.Default.Name(redisChannelSuffix) + "ws-1"
+	ps := b.client.Subscribe(context.Background(), channel)
+	defer func() { _ = ps.Close() }()
+	if _, err := ps.Receive(context.Background()); err != nil {
+		t.Fatalf("subscribe: %v", err)
+	}
+	incoming := ps.Channel()
+
+	body, _ := json.Marshal(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
+	keys := []string{
+		redisns.Default.Name(redisSeqSuffix),
+		channel,
+		redisns.Default.Name(redisEpochSuffix),
+		redisns.Default.Name(redisDedupeSuffix) + "fixed-token",
+	}
+
+	first, err := publishScript.Run(b.ctx, b.client, keys, string(body), "epoch-candidate", redisDedupeTTLSeconds).Int64()
+	if err != nil {
+		t.Fatalf("first run: %v", err)
+	}
+	if first != 1 {
+		t.Fatalf("first run must assign id 1, got %d", first)
+	}
+
+	second, err := publishScript.Run(b.ctx, b.client, keys, string(body), "epoch-candidate", redisDedupeTTLSeconds).Int64()
+	if err != nil {
+		t.Fatalf("retry: %v", err)
+	}
+	if second != 0 {
+		t.Fatalf("a retry carrying the same token must decline, got id %d", second)
+	}
+	select {
+	case <-incoming:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for the first publish")
+	}
+	select {
+	case msg := <-incoming:
+		t.Fatalf("a retried publish must reach the channel once; a second message arrived: %q", msg.Payload)
+	case <-time.After(150 * time.Millisecond):
+	}
+	// And it must not have burned an id either, or the sequence develops holes
+	// that look like lost events to every receiver.
+	got, err := mr.Get(redisns.Default.Name(redisSeqSuffix))
+	if err != nil {
+		t.Fatalf("read counter: %v", err)
+	}
+	if got != "1" {
+		t.Fatalf("a declined retry must not INCR the counter; counter reads %q", got)
+	}
+}
+
+// --- reconciliation on the receive path ---------------------------------
+
+// liveGen subscribes the workspace and returns the generation a real receive
+// loop would carry, so these tests drive fanOut the way a message does rather
+// than through the anySubscription escape hatch.
+func liveGen(t *testing.T, b *RedisBus, workspaceID string) (chan Event, int64) {
+	t.Helper()
+	ch := b.Subscribe(workspaceID)
+	t.Cleanup(func() { b.Unsubscribe(ch) })
+	return ch, b.currentSubGen(workspaceID)
+}
+
+func TestAdoptingAnEpochOntoEmptyBuffersIsNotAReset(t *testing.T) {
+	// Deliberate: otherwise every instance reports a reset at startup and
+	// pad_event_sequence_resets_total grows a per-deploy baseline, which is
+	// the thing that makes the counter unreadable.
+	b := newTestRedisBus(t)
+	obs := &recordingObserver{}
+	b.SetObserver(obs)
+
+	_, gen := liveGen(t, b, "ws-1")
+	b.fanOutFromRedis(gen, "epoch-a", Event{ID: 10, Type: ItemUpdated, WorkspaceID: "ws-1"})
+
+	if _, resets := obs.snapshot(); len(resets) != 0 {
+		t.Fatalf("learning an epoch on an empty bus must not report a reset, got %v", resets)
+	}
+	// And coverage is established from that first id.
+	if got := b.EventsSince("ws-1", 10); got == nil {
+		t.Fatal("the adopting message must establish coverage")
+	}
+}
+
+func TestAdoptingAnEpochOntoBufferedEventsIsAReset(t *testing.T) {
+	// The phase-2 roll's dangerous shape: this instance buffered BARE events
+	// from a phase-1 publisher, the counter was then reset, and a flipped
+	// publisher's first prefixed message arrives with an id ABOVE our
+	// high-water mark. The numeric check sees an ordinary successor; only the
+	// epoch says the space changed.
+	b := newTestRedisBus(t)
+	obs := &recordingObserver{}
+	b.SetObserver(obs)
+
+	_, gen := liveGen(t, b, "ws-1")
+	b.fanOutFromRedis(gen, "", Event{ID: 5, Type: ItemUpdated, WorkspaceID: "ws-1"})
+	if got := b.EventsSince("ws-1", 5); got == nil {
+		t.Fatal("fixture: the bare event must establish coverage first")
+	}
+
+	b.fanOutFromRedis(gen, "epoch-a", Event{ID: 6, Type: ItemUpdated, WorkspaceID: "ws-1"})
+
+	_, resets := obs.snapshot()
+	if len(resets) != 1 || resets[0] != ResetReasonEpochChange {
+		t.Fatalf("want one %s, got %v", ResetReasonEpochChange, resets)
+	}
+	// The pre-reset cursor must now be refused: 6 did NOT follow 5.
+	if got := b.EventsSince("ws-1", 5); got != nil {
+		t.Fatalf("a cursor from the discarded space must be a gap, got %d events", len(got))
+	}
+	// Control: a cursor at the new space's first id is served.
+	if got := b.EventsSince("ws-1", 6); got == nil {
+		t.Fatal("a cursor at the new space's first id must be served")
+	}
+}
+
+func TestAnEpochChangeDropsEveryWorkspaceAndClearsTheFloor(t *testing.T) {
+	// The counter is global, so a reset invalidates every workspace at once —
+	// and the floor is a SAME-SPACE device, so it must not survive into a
+	// space where its numbers mean nothing (that combination is a resync loop).
+	b := newTestRedisBus(t)
+	_, gen1 := liveGen(t, b, "ws-1")
+	_, gen2 := liveGen(t, b, "ws-2")
+
+	b.fanOutFromRedis(gen1, "epoch-a", Event{ID: 100, Type: ItemUpdated, WorkspaceID: "ws-1"})
+	b.fanOutFromRedis(gen2, "epoch-a", Event{ID: 101, Type: ItemUpdated, WorkspaceID: "ws-2"})
+
+	// A backwards id inside epoch-a raises the floor to 101.
+	b.fanOutFromRedis(gen1, "epoch-a", Event{ID: 50, Type: ItemUpdated, WorkspaceID: "ws-1"})
+	if b.discardedHighWater != 101 {
+		t.Fatalf("fixture: floor should stand at 101, got %d", b.discardedHighWater)
+	}
+
+	// Now the space itself changes.
+	b.fanOutFromRedis(gen1, "epoch-b", Event{ID: 1, Type: ItemUpdated, WorkspaceID: "ws-1"})
+
+	if b.discardedHighWater != 0 {
+		t.Fatalf("an epoch change must clear the floor, got %d", b.discardedHighWater)
+	}
+	if _, ok := b.replayBuffers["ws-2"]; ok {
+		t.Fatal("an epoch change must drop every workspace's buffer, not just the arriving one")
+	}
+	// The new space's first id is servable — which it would NOT be if the
+	// floor had survived, because 1 is far below 101. That is the resync loop
+	// this clears.
+	if got := b.EventsSince("ws-1", 1); got == nil {
+		t.Fatal("the new space's first id must be servable; a surviving floor is a resync loop")
+	}
+}
+
+func TestACounterGoingBackwardsRaisesTheFloor(t *testing.T) {
+	// The mixed-VERSION ordering case: a phase-1 publisher assigns and
+	// publishes in two calls, so a lower id can land after a higher one. The
+	// events between the two are unrecoverable here, so every cursor at or
+	// below what we discarded is refused.
+	b := newTestRedisBus(t)
+	obs := &recordingObserver{}
+	b.SetObserver(obs)
+	_, gen := liveGen(t, b, "ws-1")
+
+	b.fanOutFromRedis(gen, "", Event{ID: 200, Type: ItemUpdated, WorkspaceID: "ws-1"})
+	b.fanOutFromRedis(gen, "", Event{ID: 150, Type: ItemUpdated, WorkspaceID: "ws-1"})
+
+	_, resets := obs.snapshot()
+	if len(resets) != 1 || resets[0] != ResetReasonCounterBackward {
+		t.Fatalf("want one %s, got %v", ResetReasonCounterBackward, resets)
+	}
+	for _, cursor := range []int64{149, 150, 199, 200} {
+		if got := b.EventsSince("ws-1", cursor); got != nil {
+			t.Fatalf("cursor %d is at or below the discarded high-water mark 200 and must be a gap, got %d events", cursor, len(got))
+		}
+	}
+	// Control: once the sequence climbs past the discarded mark, resumes work
+	// again — the refusal is bounded, not permanent.
+	b.fanOutFromRedis(gen, "", Event{ID: 201, Type: ItemUpdated, WorkspaceID: "ws-1"})
+	if got := b.EventsSince("ws-1", 201); got == nil {
+		t.Fatal("a cursor above the discarded high-water mark must be served")
+	}
+}
+
+func TestABarePayloadAfterAdoptionLeavesTheEpochAlone(t *testing.T) {
+	// Silence is not evidence of a change. During the phase-2 roll a flipped
+	// and an un-flipped publisher both feed this instance; if the bare form
+	// were read as "epoch changed to empty", every alternating message would
+	// drop every buffer.
+	b := newTestRedisBus(t)
+	obs := &recordingObserver{}
+	b.SetObserver(obs)
+	_, gen := liveGen(t, b, "ws-1")
+
+	b.fanOutFromRedis(gen, "epoch-a", Event{ID: 10, Type: ItemUpdated, WorkspaceID: "ws-1"})
+	b.fanOutFromRedis(gen, "", Event{ID: 11, Type: ItemUpdated, WorkspaceID: "ws-1"})
+	b.fanOutFromRedis(gen, "epoch-a", Event{ID: 12, Type: ItemUpdated, WorkspaceID: "ws-1"})
+
+	if _, resets := obs.snapshot(); len(resets) != 0 {
+		t.Fatalf("a mixed-format stream inside one epoch must report no resets, got %v", resets)
+	}
+	if got := b.EventsSince("ws-1", 10); got == nil || len(got) != 2 {
+		t.Fatalf("coverage must span the mixed-format run, got %v", got)
+	}
+	if b.epoch != "epoch-a" {
+		t.Fatalf("the adopted epoch must survive a bare message, got %q", b.epoch)
+	}
+}

internal/events/redis_namespace_test.go

@@ -1,6 +1,7 @@
 package events
 
 import (
+	"strings"
 	"testing"
 	"time"
 
@@ -26,7 +27,7 @@ func TestRedisBusHonoursTheNamespace(t *testing.T) {
 	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
 	t.Cleanup(func() { _ = client.Close() })
 
-	b := NewRedisBusWithKeys(client, keys)
+	b := NewRedisBusWithKeys(client, keys, false)
 	t.Cleanup(b.Close)
 
 	// A local subscriber is what starts the Redis-side subscription for
@@ -49,6 +50,43 @@ func TestRedisBusHonoursTheNamespace(t *testing.T) {
 		t.Errorf("the namespaced bus also wrote the DEFAULT counter pad:event_seq")
 	}
 
+	// BUG-2736 added two more keys to this keyspace, and a key that misses the
+	// namespace is exactly the cross-feed BUG-2724 exists to stop — the epoch
+	// especially, since two installations sharing one epoch key would each
+	// read the other's ID-space changes as their own and drop their buffers.
+	// Driven through the flipped publish path, because that is the only path
+	// that writes them.
+	flipped := NewRedisBusWithKeys(client, keys, true)
+	t.Cleanup(flipped.Close)
+	flipped.Publish(Event{Type: "item.created", WorkspaceID: "ws-1"})
+
+	deadline = time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) && !mr.Exists("pad:inst-b:event_epoch") {
+		time.Sleep(2 * time.Millisecond)
+	}
+	if !mr.Exists("pad:inst-b:event_epoch") {
+		t.Errorf("namespaced epoch pad:inst-b:event_epoch does not exist; keys present: %v", mr.Keys())
+	}
+	if mr.Exists("pad:event_epoch") {
+		t.Errorf("the namespaced bus also wrote the DEFAULT epoch pad:event_epoch")
+	}
+	// The dedupe token carries a random suffix, so it is matched by prefix.
+	var namespacedDedupe, defaultDedupe int
+	for _, k := range mr.Keys() {
+		switch {
+		case strings.HasPrefix(k, "pad:inst-b:events:pub:"):
+			namespacedDedupe++
+		case strings.HasPrefix(k, "pad:events:pub:"):
+			defaultDedupe++
+		}
+	}
+	if namespacedDedupe == 0 {
+		t.Errorf("no namespaced dedupe token was written; keys present: %v", mr.Keys())
+	}
+	if defaultDedupe != 0 {
+		t.Errorf("the namespaced bus wrote %d dedupe tokens under the DEFAULT prefix", defaultDedupe)
+	}
+
 	waitForSubscribers(t, mr, "pad:inst-b:events:ws-1", true)
 	waitForSubscribers(t, mr, "pad:events:ws-1", false)
 

internal/events/redis_reconnect_test.go

@@ -276,7 +276,7 @@ func TestAStaleGoroutineCannotDropTheReplacementBuffer(t *testing.T) {
 	if newGen == oldGen {
 		t.Fatal("fixture drifted: resubscribing must produce a new generation")
 	}
-	b.fanOutFromRedis(newGen, Event{ID: 500, Type: ItemUpdated, WorkspaceID: "ws-1"})
+	b.fanOutFromRedis(newGen, "", Event{ID: 500, Type: ItemUpdated, WorkspaceID: "ws-1"})
 
 	// NOW the old goroutine finally notices its connection died.
 	b.dropWorkspaceCoverage("ws-1", ResetReasonSubscriptionResumed, oldGen)

internal/events/redis_resume_coverage_test.go

@@ -121,7 +121,7 @@ func TestAStragglerFromAnEndedSubscriptionCannotVouchForTheNewOne(t *testing.T)
 
 	ch := b.Subscribe("ws-1")
 	oldGen := b.currentSubGen("ws-1")
-	b.fanOutFromRedis(oldGen, Event{ID: 100, Type: ItemUpdated, WorkspaceID: "ws-1"})
+	b.fanOutFromRedis(oldGen, "", Event{ID: 100, Type: ItemUpdated, WorkspaceID: "ws-1"})
 
 	// Everyone leaves; ids 101..500 are published elsewhere, unseen.
 	b.Unsubscribe(ch)
@@ -135,14 +135,14 @@ func TestAStragglerFromAnEndedSubscriptionCannotVouchForTheNewOne(t *testing.T)
 	}
 
 	// NOW the old receive goroutine's in-flight message lands.
-	b.fanOutFromRedis(oldGen, Event{ID: 101, Type: ItemUpdated, WorkspaceID: "ws-1"})
+	b.fanOutFromRedis(oldGen, "", Event{ID: 101, Type: ItemUpdated, WorkspaceID: "ws-1"})
 
 	if got := b.EventsSince("ws-1", 101); got != nil {
 		t.Fatalf("a message from an ended subscription must not establish coverage for the new one; got %d events", len(got))
 	}
 
 	// Control: a message on the CURRENT subscription does establish coverage.
-	b.fanOutFromRedis(newGen, Event{ID: 600, Type: ItemUpdated, WorkspaceID: "ws-1"})
+	b.fanOutFromRedis(newGen, "", Event{ID: 600, Type: ItemUpdated, WorkspaceID: "ws-1"})
 	got := b.EventsSince("ws-1", 600)
 	if got == nil {
 		t.Fatal("a message on the live subscription must establish coverage")

internal/watchevents/redis_bus.go

@@ -54,6 +54,15 @@ const (
 	// question is not "is this number bigger" but "is this the same
 	// sequence". Minted once per id space by the publish script and carried
 	// on every message.
+	//
+	// AN OPAQUE TOKEN HERE, A NUMERIC BASE ON MemoryBus (BUG-2736), and the
+	// two are NOT interchangeable spellings of one idea. This counter is
+	// shared across processes, so no instance can compute an identity the
+	// others would agree with and the identity must travel with the message.
+	// MemoryBus is the sole publisher into its own space, so it can put the
+	// identity in the id's VALUE — which additionally makes a
+	// cross-incarnation cursor numerically refusable, something this epoch
+	// cannot do. See internal/idspace and MemoryBus's base field.
 	redisWatchEpochSuffix = "watchevents_epoch"
 
 	// DEPLOYMENT SCOPING (BUG-2724). These names carry the installation's

internal/watchevents/watchevents.go

@@ -366,7 +366,11 @@ type MemoryBus struct {
 	mu          sync.Mutex
 	subscribers map[chan Notification]struct{}
 	// seq counts up from base, which identifies THIS incarnation of the
-	// counter (BUG-2736). Before it, seq restarted at 1 on every process
+	// counter (BUG-2736). The asymmetry with this package's RedisBus — a
+	// numeric base here, an opaque epoch key there — is deliberate and is
+	// explained on RedisBus's epoch field; the short version is that this bus
+	// is the sole publisher into its space and can compute an identity, while
+	// a shared counter cannot be identified by any single process. Before it, seq restarted at 1 on every process
 	// start and a client resuming with a cursor from a previous incarnation
 	// was replayed the NEW space's notifications as though they followed the
 	// OLD space's — the same defect internal/events carried, for the same