diff --git a/package.json b/package.json
index 23c6485..c2bdd4d 100644
--- a/package.json
+++ b/package.json
@@ -31,14 +31,14 @@
     "test:e2e": "vitest run --project e2e",
     "test:all": "vitest run",
     "typecheck": "tsc --noEmit",
-    "prepublishOnly": "pnpm run build",
+    "prepublishOnly": "pnpm run test && pnpm run build",
     "check": "ultracite check",
     "fix": "ultracite fix"
   },
   "dependencies": {
     "@standard-schema/spec": "^1.1.0",
     "@supabase/postgrest-js": "^2.0.0",
-    "@supabase/supabase-js": "^2.0.0",
+    "@supabase/supabase-js": "^2.74.0",
     "@tanstack/db": "^0.6.0",
     "@tanstack/query-core": "^5.0.0",
     "@tanstack/query-db-collection": "^1.0.0"
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index bb8a274..f09c55f 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -15,7 +15,7 @@ importers:
         specifier: ^2.0.0
         version: 2.107.0
       '@supabase/supabase-js':
-        specifier: ^2.0.0
+        specifier: ^2.74.0
         version: 2.107.0
       '@tanstack/db':
         specifier: ^0.6.0
diff --git a/src/db.ts b/src/db.ts
index 48e1a6e..50d17ac 100644
--- a/src/db.ts
+++ b/src/db.ts
@@ -184,10 +184,13 @@ export const attachSupabaseListeners = <
       "postgres_changes",
       { event: "*", schema: "public", table: tableName },
       (payload) => {
+        // Realtime events can replay or race the initial PostgREST fetch, so
+        // an "INSERT" may already be present and an "UPDATE" may not be yet.
+        // Upsert handles both directions without throwing.
         if (payload.eventType === "INSERT") {
-          collection.utils.writeInsert(payload.new)
+          collection.utils.writeUpsert(payload.new)
         } else if (payload.eventType === "UPDATE") {
-          collection.utils.writeUpdate(payload.new)
+          collection.utils.writeUpsert(payload.new)
         } else if (payload.eventType === "DELETE") {
           const id = collection.getKeyFromItem(payload.old as T)
           if (collection.has(id)) {
diff --git a/src/functions.ts b/src/functions.ts
index aa6a547..80bc68f 100644
--- a/src/functions.ts
+++ b/src/functions.ts
@@ -12,6 +12,7 @@ import {
   type UpdateMutationFnParams,
 } from "@tanstack/db"
 import type { QueryClient, QueryMeta } from "@tanstack/query-core"
+import { CLIENT_INFO, CLIENT_INFO_HEADER } from "./request-headers"
 
 const buildQuery = (
   baseQuery: PostgrestFilterBuilder<any, any, any, any>,
@@ -127,7 +128,10 @@ export const supabaseQueryFn = async (
   // console.log(tableName, parsed);
   // console.log(tableName, cursorFilters);
 
-  let baseQuery = supabase.from(tableName).select("*")
+  let baseQuery = supabase
+    .from(tableName)
+    .select("*")
+    .setHeader(CLIENT_INFO_HEADER, CLIENT_INFO)
 
   if (parsed.limit) {
     baseQuery = baseQuery.limit(parsed.limit)
@@ -170,6 +174,7 @@ export const supabaseOnInsert = async (
         .insert({
           ...mutation.modified,
         })
+        .setHeader(CLIENT_INFO_HEADER, CLIENT_INFO)
         .select()
         .single()
 
@@ -198,10 +203,13 @@ export const supabaseOnUpdate = async (
     transaction.mutations.map(async (mutation) => {
       const { original, changes } = mutation
       const { error, data } = await filter(
-        supabase.from(tableName).update({
-          ...original,
-          ...changes,
-        }),
+        supabase
+          .from(tableName)
+          .update({
+            ...original,
+            ...changes,
+          })
+          .setHeader(CLIENT_INFO_HEADER, CLIENT_INFO),
         mutation.original
       )
         .select()
@@ -230,7 +238,10 @@ export const supabaseOnDelete = async (
   await Promise.all(
     transaction.mutations.map(async (mutation) => {
       const { error } = await filter(
-        supabase.from(tableName).delete(),
+        supabase
+          .from(tableName)
+          .delete()
+          .setHeader(CLIENT_INFO_HEADER, CLIENT_INFO),
         mutation.original
       )
 
diff --git a/src/query-once.ts b/src/query-once.ts
index 1ef5a7a..bd37f94 100644
--- a/src/query-once.ts
+++ b/src/query-once.ts
@@ -9,6 +9,7 @@ import {
   type QueryBuilder,
   queryOnce as queryOnceBase,
 } from "@tanstack/db"
+import { CLIENT_INFO, CLIENT_INFO_HEADER } from "./request-headers"
 import {
   type SerializedExpression,
   type SerializedFrom,
@@ -515,7 +516,10 @@ export function buildSupabaseQuery(
   const allWheres = [...subqueryWheres, ...(ir.where ?? [])]
 
   const selectString = buildSelectString(ir)
-  let query: SupabaseQuery = supabase.from(tableName).select(selectString)
+  let query: SupabaseQuery = supabase
+    .from(tableName)
+    .select(selectString)
+    .setHeader(CLIENT_INFO_HEADER, CLIENT_INFO)
 
   // Apply pushable where filters (skip residual / client-side filters)
   for (const w of allWheres) {
diff --git a/src/request-headers.ts b/src/request-headers.ts
new file mode 100644
index 0000000..51b5e70
--- /dev/null
+++ b/src/request-headers.ts
@@ -0,0 +1,13 @@
+import { VERSION } from "./version"
+
+/**
+ * Header name and value used to identify requests made by this library.
+ *
+ * Chain `.setHeader(CLIENT_INFO_HEADER, CLIENT_INFO)` onto a
+ * `PostgrestQueryBuilder`/`PostgrestFilterBuilder` call to stamp the
+ * request. This intentionally REPLACES supabase-js's own default
+ * `X-Client-Info` header (matching the `@supabase/ssr` convention) rather
+ * than appending to or preserving it.
+ */
+export const CLIENT_INFO_HEADER = "X-Client-Info"
+export const CLIENT_INFO = `@supabase-labs/tanstack-db/${VERSION}`
diff --git a/src/version.ts b/src/version.ts
new file mode 100644
index 0000000..2dc6a32
--- /dev/null
+++ b/src/version.ts
@@ -0,0 +1 @@
+export const VERSION = "0.0.1"
diff --git a/tests/query-once.test.ts b/tests/query-once.test.ts
index bc72d21..b6b51aa 100644
--- a/tests/query-once.test.ts
+++ b/tests/query-once.test.ts
@@ -22,14 +22,16 @@ import {
   sum,
   upper,
 } from "@tanstack/db"
-import { afterEach, beforeEach, describe, test } from "vitest"
+import { afterEach, beforeEach, describe, expect, test } from "vitest"
 import { queryOnce } from "../src/index"
+import { VERSION } from "../src/version"
 import {
   createMockedTodosCollection,
   createMockedUsersCollection,
   createMockedUsersTodosCollection,
   createMockFetch,
   expectFetchUrls,
+  getRequestHeaders,
   SUPABASE_KEY,
   SUPABASE_URL,
 } from "./test.utils"
@@ -64,6 +66,28 @@ describe("queryOnce PostgREST query generation", () => {
     })
   })
 
+  describe("automatic X-Client-Info header", () => {
+    test("is present on the request", async () => {
+      await queryOnce((q) => q.from({ user: usersCollection }), supabase)
+      expect(getRequestHeaders(mockFetch, 0).get("x-client-info")).toBe(
+        `@supabase-labs/tanstack-db/${VERSION}`
+      )
+    })
+
+    test("is present on the aggregate/groupBy PostgREST push-down path", async () => {
+      await queryOnce(
+        (q) =>
+          q
+            .from({ user: usersCollection })
+            .select(({ user }) => ({ totalUsers: count(user.id) })),
+        supabase
+      )
+      expect(getRequestHeaders(mockFetch, 0).get("x-client-info")).toBe(
+        `@supabase-labs/tanstack-db/${VERSION}`
+      )
+    })
+  })
+
   describe("WHERE", () => {
     test("WHERE active = true", async () => {
       await queryOnce(
diff --git a/tests/realtime-listeners.test.ts b/tests/realtime-listeners.test.ts
new file mode 100644
index 0000000..7d1908c
--- /dev/null
+++ b/tests/realtime-listeners.test.ts
@@ -0,0 +1,104 @@
+import type { SupabaseClient } from "@supabase/supabase-js"
+import { afterEach, describe, expect, test } from "vitest"
+import { attachSupabaseListeners } from "../src/db"
+import {
+  createMockedUsersCollection,
+  createMockFetch,
+  queryResult,
+} from "./test.utils"
+
+type RealtimePayload = any
+
+describe("attachSupabaseListeners", () => {
+  let collection: ReturnType<typeof createMockedUsersCollection>
+
+  afterEach(() => {
+    collection?.cleanup()
+  })
+
+  async function setup() {
+    const captured: Array<(payload: RealtimePayload) => void> = []
+    const fakeSupabase = {
+      channel: () => {
+        const stub = {
+          on: (
+            _event: string,
+            _filter: unknown,
+            cb: (payload: RealtimePayload) => void
+          ) => {
+            captured.push(cb)
+            return stub
+          },
+          subscribe: () => stub,
+        }
+        return stub
+      },
+    } as unknown as SupabaseClient
+
+    const mockFetch = createMockFetch()
+    collection = createMockedUsersCollection(mockFetch)
+    // Preload from the initial PostgREST fetch, mirroring how the collection
+    // is populated before any realtime event can race or replay against it.
+    await queryResult((q) => q.from({ user: collection }))
+
+    attachSupabaseListeners(fakeSupabase, "users", collection)
+
+    const [handler] = captured
+    if (!handler) {
+      throw new Error("expected attachSupabaseListeners to register a callback")
+    }
+    return handler
+  }
+
+  test("INSERT for an already-present key with changed data does not throw", async () => {
+    const handler = await setup()
+    // Establish a row as already-present via a realtime INSERT, matching how
+    // a real row (numeric id, schema-valid) would already be synced.
+    const row = { id: 501, name: "Dana", email: "dana@test.com", active: true }
+    handler({ eventType: "INSERT", new: row })
+    expect(collection.get(String(row.id))?.name).toBe("Dana")
+
+    // Realtime replayed/duplicated the same INSERT with changed data: this is
+    // the exact shape of the unhandled `CollectionOperationError` from CI.
+    expect(() =>
+      handler({
+        eventType: "INSERT",
+        new: { ...row, name: "Dana Prime" },
+      })
+    ).not.toThrow()
+
+    expect(collection.get(String(row.id))?.name).toBe("Dana Prime")
+  })
+
+  test("UPDATE for a key not present in the collection does not throw", async () => {
+    const handler = await setup()
+    const newRow = {
+      id: 777,
+      name: "Charlie",
+      email: "charlie@test.com",
+      active: true,
+    }
+
+    expect(() =>
+      handler({
+        eventType: "UPDATE",
+        new: newRow,
+      })
+    ).not.toThrow()
+
+    expect(collection.get(String(newRow.id))?.name).toBe("Charlie")
+  })
+
+  test("DELETE for a key not present does not throw", async () => {
+    const handler = await setup()
+
+    expect(() =>
+      handler({
+        eventType: "DELETE",
+        old: { id: "user_missing" },
+      })
+    ).not.toThrow()
+
+    expect(collection.has("user_missing")).toBe(false)
+  })
+})
diff --git a/tests/request-headers.test.ts b/tests/request-headers.test.ts
new file mode 100644
index 0000000..b3faaa9
--- /dev/null
+++ b/tests/request-headers.test.ts
@@ -0,0 +1,139 @@
+import { createClient } from "@supabase/supabase-js"
+import { createCollection } from "@tanstack/db"
+import { afterEach, describe, expect, test } from "vitest"
+import pkg from "../package.json"
+import { supabaseCollectionOptions } from "../src/index"
+import { VERSION } from "../src/version"
+import {
+  createMockedUsersCollection,
+  createMockFetch,
+  getRequestHeaders,
+  queryResult,
+  SUPABASE_KEY,
+  SUPABASE_URL,
+  usersSchema,
+} from "./test.utils"
+
+const EXPECTED_CLIENT_INFO = `@supabase-labs/tanstack-db/${VERSION}`
+
+test("VERSION stays in sync with package.json", () => {
+  expect(VERSION).toBe(pkg.version)
+})
+
+describe("automatic X-Client-Info header", () => {
+  let collection: ReturnType<typeof createMockedUsersCollection>
+
+  afterEach(() => {
+    collection?.cleanup()
+  })
+
+  test("is present on select requests", async () => {
+    const mockFetch = createMockFetch()
+    collection = createMockedUsersCollection(mockFetch)
+
+    await queryResult((q) => q.from({ user: collection }))
+
+    expect(getRequestHeaders(mockFetch, 0).get("x-client-info")).toBe(
+      EXPECTED_CLIENT_INFO
+    )
+  })
+
+  test("is present on insert requests", async () => {
+    const mockFetch = createMockFetch()
+    collection = createMockedUsersCollection(mockFetch)
+    await queryResult((q) => q.from({ user: collection }))
+
+    const tx = collection.insert({
+      id: 2,
+      name: "Bob",
+      email: "bob@test.com",
+      active: true,
+    })
+    await tx.isPersisted.promise
+
+    const lastCall = mockFetch.mock.calls.length - 1
+    expect(getRequestHeaders(mockFetch, lastCall).get("x-client-info")).toBe(
+      EXPECTED_CLIENT_INFO
+    )
+  })
+
+  test("is present on update requests", async () => {
+    const mockFetch = createMockFetch()
+    collection = createMockedUsersCollection(mockFetch)
+    await queryResult((q) => q.from({ user: collection }))
+
+    const insertTx = collection.insert({
+      id: 2,
+      name: "Bob",
+      email: "bob@test.com",
+      active: true,
+    })
+    await insertTx.isPersisted.promise
+
+    const updateTx = collection.update("2", (draft) => {
+      draft.name = "Bobby"
+    })
+    await updateTx.isPersisted.promise
+
+    const lastCall = mockFetch.mock.calls.length - 1
+    expect(getRequestHeaders(mockFetch, lastCall).get("x-client-info")).toBe(
+      EXPECTED_CLIENT_INFO
+    )
+  })
+
+  test("is present on delete requests", async () => {
+    const mockFetch = createMockFetch()
+    collection = createMockedUsersCollection(mockFetch)
+    await queryResult((q) => q.from({ user: collection }))
+
+    const insertTx = collection.insert({
+      id: 2,
+      name: "Bob",
+      email: "bob@test.com",
+      active: true,
+    })
+    await insertTx.isPersisted.promise
+
+    const deleteTx = collection.delete("2")
+    await deleteTx.isPersisted.promise
+
+    const lastCall = mockFetch.mock.calls.length - 1
+    expect(getRequestHeaders(mockFetch, lastCall).get("x-client-info")).toBe(
+      EXPECTED_CLIENT_INFO
+    )
+  })
+
+  test("replaces supabase-js's own X-Client-Info rather than appending", async () => {
+    const mockFetch = createMockFetch()
+    collection = createMockedUsersCollection(mockFetch)
+
+    await queryResult((q) => q.from({ user: collection }))
+
+    const value = getRequestHeaders(mockFetch, 0).get("x-client-info")
+    expect(value).toBe(EXPECTED_CLIENT_INFO)
+    expect(value).not.toContain("supabase-js")
+  })
+
+  test("preserves user-supplied global headers", async () => {
+    const mockFetch = createMockFetch()
+    collection = createCollection(
+      supabaseCollectionOptions({
+        tableName: "users",
+        keys: ["id"],
+        schema: usersSchema,
+        supabase: createClient(SUPABASE_URL, SUPABASE_KEY, {
+          global: {
+            fetch: mockFetch,
+            headers: { "X-Custom-Header": "custom-value" },
+          },
+        }),
+      })
+    )
+
+    await queryResult((q) => q.from({ user: collection }))
+
+    const headers = getRequestHeaders(mockFetch, 0)
+    expect(headers.get("x-custom-header")).toBe("custom-value")
+    expect(headers.get("x-client-info")).toBe(EXPECTED_CLIENT_INFO)
+  })
+})
diff --git a/tests/test.utils.ts b/tests/test.utils.ts
index 1a53561..51d43ed 100644
--- a/tests/test.utils.ts
+++ b/tests/test.utils.ts
@@ -46,9 +46,24 @@ export const mockResponses: Record<string, any[]> = {
 }
 
 export function createMockFetch() {
-  return vi.fn<typeof fetch>().mockImplementation((input) => {
+  return vi.fn<typeof fetch>().mockImplementation((input, init) => {
     const url = new URL(typeof input === "string" ? input : input.toString())
     const table = url.pathname.replace("/rest/v1/", "")
+    const method = init?.method ?? "GET"
+
+    // For insert/update requests, echo the request body back as the
+    // "representation" so `.select().single()` has something schema-shaped
+    // to parse, mirroring what PostgREST would return.
+    if (method === "POST" || method === "PATCH") {
+      const body = init?.body ? JSON.parse(init.body as string) : {}
+      return Promise.resolve(
+        new Response(JSON.stringify(body), {
+          status: 200,
+          headers: { "content-type": "application/json" },
+        })
+      )
+    }
+
     const response = mockResponses[table] ?? []
     return Promise.resolve(
       new Response(JSON.stringify(response), {
@@ -59,6 +74,15 @@ export function createMockFetch() {
   })
 }
 
+/** Extract the headers of a captured `fetch` call as a `Headers` instance. */
+export function getRequestHeaders(
+  mockFetch: ReturnType<typeof createMockFetch>,
+  callIndex = 0
+): Headers {
+  const init = mockFetch.mock.calls[callIndex]?.[1]
+  return new Headers(init?.headers as ConstructorParameters<typeof Headers>[0])
+}
+
 export function createMockedUsersCollection(mockFetch: typeof fetch) {
   return createCollection(
     supabaseCollectionOptions({