Commit and publish are two instants

const committed = await api.sendMessage(connection.identity, { … });
…                                    // the ack, then nineteen lines of comment
…
if (!committed.duplicate && committed.text !== null) {
  await fanout?.publish({ id: committed.id, channel: committed.channel_id, seq: committed.seq, … });
}

@@ -191,7 +191,123 @@ describe("resume across a real fabric", () => {
     // the phase went back to normal 2.6 delivery.
     await publishFromElsewhere(frame(44));
     await settle(300);
     expect(created(frames)).toEqual([42, 44]);
     socket.close();
   });
+
+  it("suppresses a frame the backfill already delivered, published after the resume", async () => {
+    // THE FOURTH QUADRANT, and the defect this chapter exists to close.
+    //
+    // The three tests above cover: published while buffering and in the backfill
+    // (deduplicated by `flushable`); published while buffering and NOT in the
+    // backfill (delivered by the flush); published after going live with a
+    // sequence ABOVE the mark (delivered). The cell they leave empty is this one —
+    // published after going live, with a sequence the backfill already sent.
+    //
+    // It is not a contrived case. A message is durable at one instant and
+    // announced at another: the gateway commits through the api and only then
+    // publishes to Redis. A backfill query landing between those two instants
+    // returns a message the fabric has not yet delivered, and the delivery arrives
+    // after the resume has finished — when chapter 2.7's dedup window has already
+    // closed, because `marks` was a local variable that `resume()` discarded.
+    //
+    // One number different from the test above it. That is the whole bug.
+    harness = await boot({
+      session: async () => ({
+        environment_id: "env-1",
+        user: "tuan",
+        channel_ids: [CHANNEL],
+      }),
+      backfill: async () => ({
+        [CHANNEL]: { messages: [frame(42)], truncated: false },
+      }),
+      sendMessage: async () => {
+        throw new Error("not used");
+      },
+    });
+    const socket = new WebSocket(
+      `${harness.url}?token=${await token()}&cursor=${CHANNEL}:41`,
+    );
+    const frames = record(socket);
+    await settle(400);
+    // The resume has completed. NOW the fabric catches up with a message the
+    // backfill already delivered — the publish that was still in flight while the
+    // backfill query ran.
+    await publishFromElsewhere(frame(42));
+    await settle(300);
+
+    expect(created(frames)).toEqual([42]);
+    socket.close();
+  });
+
+  it("still suppresses when two instances publish out of order", async () => {
+    // Sequences COMMIT in order under a channel row lock; they are PUBLISHED by
+    // whichever gateway instance handled each send, and those do not coordinate.
+    // A prompt publish of 43 can precede a stalled publish of 42.
+    //
+    // This is the case that made the spec's first design unsafe. It proposed
+    // retiring the mark once a higher sequence arrived — which would see the 43,
+    // drop the mark, and then deliver the 42 (research R3).
+    harness = await boot({
+      session: async () => ({
+        environment_id: "env-1",
+        user: "tuan",
+        channel_ids: [CHANNEL],
+      }),
+      backfill: async () => ({
+        [CHANNEL]: { messages: [frame(42)], truncated: false },
+      }),
+      sendMessage: async () => {
+        throw new Error("not used");
+      },
+    });
+    const socket = new WebSocket(
+      `${harness.url}?token=${await token()}&cursor=${CHANNEL}:41`,
+    );
+    const frames = record(socket);
+    await settle(400);
+    // 43 is ABOVE the mark and must be delivered. A rule that retired the mark on
+    // seeing it would then have nothing left to compare the delayed 42 against.
+    await publishFromElsewhere(frame(43));
+    await settle(150);
+    // 42 is the mark itself, arriving late from an instance that stalled between
+    // its api call and its publish. It must still be suppressed.
+    await publishFromElsewhere(frame(42));
+    await settle(300);
+
+    expect(created(frames)).toEqual([42, 43]);
+    socket.close();
+  });
+
+  it("suppresses nothing when the resume degraded", async () => {
+    // A degraded resume tells the client to page history for every channel, so the
+    // backfill it received is a fragment or nothing at all. A mark taken from it
+    // would suppress messages the client never got — turning this chapter's
+    // duplicate into a gap, which constitution II ranks worse.
+    harness = await boot({
+      session: async () => ({
+        environment_id: "env-1",
+        user: "tuan",
+        channel_ids: [CHANNEL],
+      }),
+      backfill: async () => {
+        throw new Error("backfill unavailable");
+      },
+      sendMessage: async () => {
+        throw new Error("not used");
+      },
+    });
+    const socket = new WebSocket(
+      `${harness.url}?token=${await token()}&cursor=${CHANNEL}:41`,
+    );
+    const frames = record(socket);
+    await settle(400);
+    // A sequence at or below the presented cursor. With no mark retained it must
+    // still arrive: the client was told to page history, not to expect silence.
+    await publishFromElsewhere(frame(41));
+    await settle(300);
+
+    expect(created(frames)).toEqual([41]);
+    socket.close();
+  });
 });

$ pnpm --filter @relay/gateway test:integration src/resume.itest.ts
 × suppresses a frame the backfill already delivered, published after the resume
 × still suppresses when two instances publish out of order
 Tests  2 failed | 4 passed (6)
 
AssertionError: expected [ 42, 42 ] to deeply equal [ 42 ]

@@ -104,12 +104,57 @@ export function flushable(
   buffer: Message[],
   marks: Record<string, number>,
 ): Message[] {
   return buffer.filter((frame) => frame.seq > (marks[frame.channel] ?? 0));
 }
 
+/** Step 5's half, and the reason this chapter exists (FR-RTM-03).
+ *
+ * `flushable` answers this question for frames sitting in the buffer. This answers
+ * it for frames arriving AFTER the connection has gone live, which is the case
+ * chapter 2.7 did not have — its marks were a local variable that `resume()`
+ * discarded the moment it flushed.
+ *
+ * A message is durable at one instant and announced at another: the gateway
+ * commits through the api and only then publishes to the fabric. A backfill query
+ * landing between those two instants returns a message the fabric has not yet
+ * delivered, so the delivery arrives once the resume is over — and before this
+ * function there was nothing left to compare it against.
+ *
+ * `<=`, not `<`. The mark IS a sequence the backfill delivered rather than the one
+ * after it, which is the same off-by-one `flushable` documents one screen up. */
+export function suppressed(
+  marks: Record<string, number> | null,
+  frame: Message,
+): boolean {
+  if (marks === null) return false;
+  const mark = marks[frame.channel];
+  return mark !== undefined && frame.seq <= mark;
+}
+
+/** The marks a connection keeps, bounded to the channels it actually presented
+ * cursors for.
+ *
+ * `highWaterMarks` seeds from the cursors and then adds a key for every channel
+ * the backfill answered with, so on its own it bounds nothing. The api derives its
+ * response from the cursors it was given, so the two agree today — but a bound
+ * this service claims should not live in another service's response shape. This is
+ * `scopeCursors` applied one step later, and it is here rather than in `session.ts`
+ * so that a unit test can reach it.
+ *
+ * At most `MAX_RESUME_CHANNELS` entries, because that is what the resume contract
+ * already caps the cursor map at. */
+export function scopeMarks(
+  marks: Record<string, number>,
+  cursors: Record<string, number>,
+): Record<string, number> {
+  return Object.fromEntries(
+    Object.entries(marks).filter(([channelId]) => channelId in cursors),
+  );
+}
+
 /** A promise that resolves false instead of hanging forever. */
 export async function withDeadline(
   work: Promise<unknown>,
   ms: number,
 ): Promise<boolean> {
   let timer: ReturnType<typeof setTimeout> | undefined;

@@ -26,12 +26,30 @@ export interface Connection {
    * field and nothing else — the resume machinery is invisible to it. */
   phase: ResumePhase;
   buffer: Message[];
   /** Set when the buffer hit its ceiling. The frames are gone, so the
    * client must be told to page history instead of trusting the stream. */
   overflowed: boolean;
+  /** Per channel, the highest sequence this connection's backfill
+   * delivered — kept for the connection's life so that a frame the fabric
+   * announces AFTER the resume has finished can still be recognised as one the
+   * client already holds.
+   *
+   * Three states, and the middle one is not the same as the first:
+   *
+   *   null            a fresh connect, or a resume that degraded — suppress nothing
+   *   {}              a resume whose cursor set was empty after scoping
+   *   { chan: 42 }    suppress at or below 42 on that channel
+   *
+   * NEVER RETIRED while the connection lives. Retiring on a higher sequence looks
+   * like the natural way to bound this and hands the duplicate straight back:
+   * sequences commit in order under a channel row lock but are published by
+   * whichever gateway instance handled each send, so a prompt 43 can beat a stalled
+   * 42. Bounded instead by `MAX_RESUME_CHANNELS`, which already caps the cursors
+   * these are scoped to. */
+  marks: Record<string, number> | null;
 }
 
 export class Registry {
   private readonly byId = new Map<string, Connection>();
 
   add(connection: Connection): void {

@@ -11,12 +11,14 @@ import type { Fanout } from "./fanout.js";
 import { Registry, type Connection } from "./registry.js";
 import {
   MAX_BUFFERED_FRAMES,
   SUBSCRIBE_DEADLINE_MS,
   flushable,
   highWaterMarks,
+  scopeMarks,
+  suppressed,
   parseCursors,
   scopeCursors,
   withDeadline,
 } from "./resume.js";
 
 // One session per socket (chapter 2.5). The order of operations here is the
@@ -86,12 +88,19 @@ export function attachSessions({
           connection.overflowed = true;
           continue;
         }
         connection.buffer.push(message);
         continue;
       }
+      // A live connection is not necessarily a connection with
+      // nothing to remember: a frame at or below what its backfill already
+      // delivered is one it has, however long ago the resume finished. Before
+      // this, delivery consulted `phase` and nothing else, and the marks were
+      // discarded the moment the connection went live — which is precisely when
+      // the fabric could still be catching up.
+      if (suppressed(connection.marks, message)) continue;
       send(connection.socket, { type: "message.created", payload: message });
     }
   }
   fanout?.onDelivery(deliver);
   // noServer: the upgrade is handled by hand so the token can be checked
   // BEFORE the handshake completes. Letting ws own the upgrade would mean
@@ -153,12 +162,15 @@ export function attachSessions({
       // questions at once, and a failure now closes the socket before it opens.
       channelIds: new Set(channelIds),
       missedPings: 0,
       phase: presented === undefined ? "live" : "buffering",
       buffer: [],
       overflowed: false,
+      // A fresh connect suppresses nothing; a resume fills this in when it
+      // succeeds, and leaves it null when it degrades.
+      marks: null,
     };
 
     registry.add(connection);
     // Subscriptions follow membership: the first local member of a channel
     // makes this instance a subscriber, and the last one to leave releases
     // it (reference-counted in the fabric).
@@ -255,12 +267,17 @@ export function attachSessions({
     /** Everything that cannot promise completeness ends up here: the client
      * is told resume did not happen and which channels to page instead. The
      * frames held so far are dropped on purpose — they would be an arbitrary
      * fragment of a stream the client is about to refetch in full. */
     const degrade = (reason: string): void => {
       connection.buffer = [];
+      // AND THE MARKS. A degraded resume tells the client to page history for
+      // every channel, so the backfill behind these marks is a fragment or
+      // nothing at all — suppressing on them would turn this chapter's duplicate
+      // into a gap, which constitution II ranks worse.
+      connection.marks = null;
       connection.phase = "live";
       ack(connection, {
         cursor: cursors,
         resume_ok: false,
         truncated: [...connection.channelIds],
       });
@@ -329,12 +346,16 @@ export function attachSessions({
       return;
     }
     for (const message of flushable(connection.buffer, marks)) {
       send(connection.socket, { type: "message.created", payload: message });
     }
     connection.buffer = [];
+    // KEPT, where chapter 2.7 discarded them. Scoped to the cursors this
+    // connection actually presented, so the bound is this service's rather than
+    // one inherited from the shape of the api's response.
+    connection.marks = scopeMarks(marks, cursors);
     // Step 5.
     connection.phase = "live";
     logger.log("info", "resume.completed", {
       connection_id: connection.id,
       backfilled: Object.values(backfilled).reduce(
         (n, page) => n + page.messages.length,

resume completes, mark = 4
seq 5 arrives  → above the mark → deliver, and RETIRE
seq 4 arrives  → no mark left   → deliver   ← the duplicate, again

@@ -4,12 +4,14 @@ import type { Message } from "@relay/protocol";
 
 import {
   flushable,
   highWaterMarks,
   parseCursors,
   scopeCursors,
+  scopeMarks,
+  suppressed,
   withDeadline,
 } from "./resume.js";
 
 // The resume theorem, held still (chapter 2.7). Everything here is a pure
 // function precisely so the ordering argument can be tested without a
 // socket, a broker, or a database — session.test.ts then proves the
@@ -131,6 +133,88 @@ describe("withDeadline", () => {
     // connection away.
     expect(await withDeadline(Promise.reject(new Error("redis down")), 50)).toBe(
       false,
     );
   });
 });
+
+// `flushable` decides what a resuming connection may hand over from
+// its buffer; `suppressed` decides what a LIVE connection must still refuse. The
+// two comparisons are the same and the second one did not exist, which is the
+// whole of the defect.
+describe("suppressed", () => {
+  const at = (channel: string, seq: number): Message => ({
+    id: `id-${seq}`,
+    channel,
+    seq,
+    user: "dispatcher",
+    text: `m${seq}`,
+    created_at: "2026-08-19T00:00:00.000Z",
+  });
+
+  it("suppresses a frame at the mark", () => {
+    // `<=`, not `<`: the mark IS a sequence the backfill delivered.
+    expect(suppressed({ a: 42 }, at("a", 42))).toBe(true);
+  });
+
+  it("suppresses a frame below the mark", () => {
+    expect(suppressed({ a: 42 }, at("a", 7))).toBe(true);
+  });
+
+  it("delivers a frame above the mark", () => {
+    // The half that matters most: suppression must never become a gap.
+    expect(suppressed({ a: 42 }, at("a", 43))).toBe(false);
+  });
+
+  it("delivers on a channel with no mark", () => {
+    expect(suppressed({ a: 42 }, at("b", 1))).toBe(false);
+  });
+
+  it("suppresses nothing when the connection never resumed", () => {
+    expect(suppressed(null, at("a", 1))).toBe(false);
+  });
+
+  it("suppresses nothing when the mark set is empty", () => {
+    expect(suppressed({}, at("a", 1))).toBe(false);
+  });
+
+  it("keeps channels apart", () => {
+    // A mark on one channel must never silence another. Two connections' worth of
+    // confusion would look exactly like message loss.
+    const marks = { a: 42, b: 1 };
+    expect(suppressed(marks, at("a", 40))).toBe(true);
+    expect(suppressed(marks, at("b", 40))).toBe(false);
+  });
+});
+
+describe("scopeMarks", () => {
+  it("drops a channel the cursors never named", () => {
+    // `highWaterMarks` adds a key for every channel the BACKFILL answered with.
+    // The api derives its response from the cursors it was given, so this cannot
+    // happen today — and a bound this service claims should not depend on another
+    // service's response shape.
+    expect(scopeMarks({ a: 42, surprise: 9 }, { a: 41 })).toEqual({ a: 42 });
+  });
+
+  it("keeps every channel the cursors did name", () => {
+    expect(scopeMarks({ a: 42, b: 7 }, { a: 41, b: 1 })).toEqual({ a: 42, b: 7 });
+  });
+
+  it("never exceeds the cursor set, which the resume contract already caps", () => {
+    const cursors = Object.fromEntries(
+      Array.from({ length: 200 }, (_, i) => [`c${i}`, i]),
+    );
+    const marks = { ...cursors, extra: 1 };
+    expect(Object.keys(scopeMarks(marks, cursors))).toHaveLength(200);
+  });
+
+  it("keeps a cursor far above anything the channel holds", () => {
+    // A client-supplied cursor is checked only for being a non-negative integer.
+    // A nonsense value seeds a mark rather than being clamped, and that is the
+    // documented consequence rather than a defect: the client asserted it holds
+    // those sequences, and the backfill has always taken that at face value.
+    // The blast radius is one connection and reconnecting recovers it.
+    expect(scopeMarks({ a: 999_999_999 }, { a: 999_999_999 })).toEqual({
+      a: 999_999_999,
+    });
+  });
+});

@@ -466,12 +466,79 @@ describe("the socket (chapter 2.5)", () => {
     const seqs = created(frames);
     expect(seqs).toEqual([42, 43]); // complete…
     expect(new Set(seqs).size).toBe(seqs.length); // …and exactly once
     socket.close();
   });
 
+  it("a frame at the mark, arriving after the resume, is not delivered", async () => {
+    // The same property `resume.itest.ts` proves against a real broker, held here
+    // against a stubbed one so it fails fast and without Redis. This is the wiring
+    // rather than the predicate: that `deliver()` consults the marks at all.
+    const fanout = stubFanout();
+    harness = await boot(
+      stubApi({
+        backfill: async () => ({
+          [CHANNEL]: { messages: [frame(42)], truncated: false },
+        }),
+      }),
+      undefined,
+      fanout,
+    );
+    const socket = new WebSocket(
+      `${harness.url}?token=${await token()}&cursor=${CHANNEL}:41`,
+    );
+    const frames = record(socket);
+    await nextFrame(socket, "connection.ack");
+    await settle();
+    // The resume is over. The fabric catches up with what the backfill sent.
+    fanout.emit(frame(42));
+    await settle();
+    expect(created(frames)).toEqual([42]);
+    socket.close();
+  });
+
+  it("a frame above the mark, arriving after the resume, IS delivered", async () => {
+    // The half that stops a duplicate fix becoming a gap (FR-RTM-03 is one
+    // property, not two: no gap AND no double).
+    const fanout = stubFanout();
+    harness = await boot(
+      stubApi({
+        backfill: async () => ({
+          [CHANNEL]: { messages: [frame(42)], truncated: false },
+        }),
+      }),
+      undefined,
+      fanout,
+    );
+    const socket = new WebSocket(
+      `${harness.url}?token=${await token()}&cursor=${CHANNEL}:41`,
+    );
+    const frames = record(socket);
+    await nextFrame(socket, "connection.ack");
+    await settle();
+    fanout.emit(frame(43));
+    await settle();
+    expect(created(frames)).toEqual([42, 43]);
+    socket.close();
+  });
+
+  it("a connection that never resumed suppresses nothing", async () => {
+    // A fresh connect presents no cursor, so it holds no marks and behaves
+    // exactly as chapter 2.6 left it.
+    const fanout = stubFanout();
+    harness = await boot(stubApi({}), undefined, fanout);
+    const socket = new WebSocket(`${harness.url}?token=${await token()}`);
+    const frames = record(socket);
+    await nextFrame(socket, "connection.ack");
+    await settle();
+    fanout.emit(frame(1));
+    await settle();
+    expect(created(frames)).toEqual([1]);
+    socket.close();
+  });
+
   it("stages the other interleaving: published during backfill, absent from it", async () => {
     // Same window, the other order: seq 43 committed AFTER the backfill
     // query's snapshot, so it exists ONLY in the buffer. This is the
     // interleaving where backfill-then-subscribe loses the message.
     const fanout = stubFanout();
     harness = await boot(

@@ -75,14 +75,24 @@ export default defineConfig({
         "services/api/src/db/repository.ts": {
           branches: 85,
           functions: 100,
           lines: 98,
           statements: 95,
         },
+        // THE DEDUPLICATION CHAPTER RAISED THIS, 93 -> 95. The chapter added two pure functions
+        // to this file — the live-path suppression predicate and the scoping that
+        // bounds the marks — and both are fully covered.
+        //
+        // Not 100, and the missing branch is named rather than chased: it is
+        // `if (timer)` in chapter 2.7's `withDeadline`, whose falsy arm cannot be
+        // reached because a Promise executor runs synchronously and always assigns
+        // the timer before the `finally` can see it. Pinning 100 here would pin a
+        // number the file cannot reach without deleting a defensive check that
+        // costs nothing.
         "services/gateway/src/resume.ts": {
-          branches: 93,
+          branches: 95,
           functions: 100,
           lines: 100,
           statements: 100,
         },
         "services/api/src/auth/user-token.ts": {
           branches: 96,

=============== mutation 1: never keep the marks after a successful resume
  RESULT: caught  × suppresses a frame the backfill already delivered…
=============== mutation 2: use < instead of <= in the predicate
  RESULT: caught  × suppresses a frame the backfill already delivered…
=============== mutation 3: suppress on any channel rather than the frame's own
  RESULT: caught  × delivers on a channel with no mark
                  × keeps channels apart
=============== mutation 4: ignore the cursors when scoping the marks
  RESULT: caught  × drops a channel the cursors never named
=============== mutation 5: retire a mark once a higher sequence is delivered
  RESULT: caught  × still suppresses when two instances publish out of order
=============== files restored byte-identical?
  YES

expected [ …(2756) ] to include '<uuid>'

@@ -424,29 +424,61 @@ describe("the consumer", () => {
 
   it("invariant 9: a consumer stopped for N publishes receives all N on restart", async () => {
     // What `limits` retention means: the stream holds messages whether or not
     // anybody is reading. The backlog waits.
     const durable = `itest-catchup-${Date.now()}`;
     const seen: string[] = [];
-    const runtime = runtimeFor(db, durable, async (e) => void seen.push(e.id));
+    // ONE environment for all three publishes, and the consumer filtered to it.
+    //
+    // This test used to call `ENV()` three times — and `ENV` mints a fresh uuid
+    // on every call, so the three events went to three different subjects and no
+    // filter could cover them. Without a filter the durable starts at the head of
+    // a stream holding ~13,000 events from earlier chapters, and the loop below
+    // had to drain all of them inside a fixed budget of 800 polls before the
+    // three under test were even reachable.
+    //
+    // Found on run 11 of the deduplication chapter's twenty post-fix lane runs: `expected
+    // [ …(2756) ] to include '<uuid>'`. 2,756 events drained and the backlog
+    // still not cleared. It is the same shape as the sweep and the drain in
+    // `deliveries.itest.ts` — a test riding a shared, growing resource with a
+    // fixed budget, which passes until the resource outgrows the budget.
+    //
+    // Scoped, the drain below has nothing to drain and the assertion is about
+    // exactly the three events it published. The invariant is unchanged: the
+    // stream holds messages whether or not anybody is reading, and the backlog
+    // waits.
+    const environmentId = ENV();
+    const runtime = runtimeFor(
+      db,
+      durable,
+      async (e) => void seen.push(e.id),
+      silent,
+      environmentId,
+    );
 
     // Get to the head of the stream first, so "everything published while away"
-    // is measurable rather than lost in twelve thousand older events.
+    // is measurable rather than lost among older events.
     for (let i = 0; i < 800; i++) {
       const { handled, duplicates } = await runtime.pollOnce();
       if (handled + duplicates === 0) break;
     }
     await runtime.stop();
 
     const published = [
-      await publish(ENV()),
-      await publish(ENV()),
-      await publish(ENV()),
+      await publish(environmentId),
+      await publish(environmentId),
+      await publish(environmentId),
     ];
 
-    const restarted = runtimeFor(db, durable, async (e) => void seen.push(e.id));
+    const restarted = runtimeFor(
+      db,
+      durable,
+      async (e) => void seen.push(e.id),
+      silent,
+      environmentId,
+    );
     for (let i = 0; i < 100; i++) {
       await restarted.pollOnce();
       if (published.every((id) => seen.includes(id))) break;
     }
     await restarted.stop();