diff --git a/.changeset/better-plums-thank.md b/.changeset/better-plums-thank.md
new file mode 100644
index 000000000..6f92eb47f
--- /dev/null
+++ b/.changeset/better-plums-thank.md
@@ -0,0 +1,7 @@
+---
+"@tanstack/db-collections": patch
+"@tanstack/db-example-react-todo": patch
+"@tanstack/db": patch
+---
+
+Type PendingMutation whenever possible
diff --git a/examples/react/todo/src/App.tsx b/examples/react/todo/src/App.tsx
index ad0b5b169..97cd7e21b 100644
--- a/examples/react/todo/src/App.tsx
+++ b/examples/react/todo/src/App.tsx
@@ -132,10 +132,10 @@ const createTodoCollection = (type: CollectionType) => {
   if (collectionsCache.has(`todo`)) {
     return collectionsCache.get(`todo`)
   } else {
-    let newCollection: Collection
+    let newCollection: Collection<UpdateTodo>
     if (type === CollectionType.Electric) {
-      newCollection = createCollection<UpdateTodo>(
-        electricCollectionOptions({
+      newCollection = createCollection(
+        electricCollectionOptions<UpdateTodo>({
           id: `todos`,
           shapeOptions: {
             url: `http://localhost:3003/v1/shape`,
@@ -237,7 +237,7 @@ const createConfigCollection = (type: CollectionType) => {
   if (collectionsCache.has(`config`)) {
     return collectionsCache.get(`config`)
   } else {
-    let newCollection: Collection
+    let newCollection: Collection<UpdateConfig>
     if (type === CollectionType.Electric) {
       newCollection = createCollection(
         electricCollectionOptions({
diff --git a/packages/db-collections/src/electric.ts b/packages/db-collections/src/electric.ts
index 005e230d7..90af0d53d 100644
--- a/packages/db-collections/src/electric.ts
+++ b/packages/db-collections/src/electric.ts
@@ -40,7 +40,9 @@ export interface ElectricCollectionConfig<T extends Row<unknown>> {
    * @param params Object containing transaction and mutation information
    * @returns Promise resolving to an object with txid
    */
-  onInsert?: (params: MutationFnParams) => Promise<{ txid: string } | undefined>
+  onInsert?: (
+    params: MutationFnParams<T>
+  ) => Promise<{ txid: string } | undefined>
 
   /**
    * Optional asynchronous handler function called before an update operation
@@ -48,7 +50,9 @@ export interface ElectricCollectionConfig<T extends Row<unknown>> {
    * @param params Object containing transaction and mutation information
    * @returns Promise resolving to an object with txid
    */
-  onUpdate?: (params: MutationFnParams) => Promise<{ txid: string } | undefined>
+  onUpdate?: (
+    params: MutationFnParams<T>
+  ) => Promise<{ txid: string } | undefined>
 
   /**
    * Optional asynchronous handler function called before a delete operation
@@ -56,10 +60,12 @@ export interface ElectricCollectionConfig<T extends Row<unknown>> {
    * @param params Object containing transaction and mutation information
    * @returns Promise resolving to an object with txid
    */
-  onDelete?: (params: MutationFnParams) => Promise<{ txid: string } | undefined>
+  onDelete?: (
+    params: MutationFnParams<T>
+  ) => Promise<{ txid: string } | undefined>
 }
 
-function isUpToDateMessage<T extends Row<unknown> = Row>(
+function isUpToDateMessage<T extends Row<unknown>>(
   message: Message<T>
 ): message is ControlMessage & { up_to_date: true } {
   return isControlMessage(message) && message.headers.control === `up-to-date`
@@ -133,7 +139,7 @@ export function electricCollectionOptions<T extends Row<unknown>>(
 
   // Create wrapper handlers for direct persistence operations that handle txid awaiting
   const wrappedOnInsert = config.onInsert
-    ? async (params: MutationFnParams) => {
+    ? async (params: MutationFnParams<T>) => {
         const handlerResult = (await config.onInsert!(params)) ?? {}
         const txid = (handlerResult as { txid?: string }).txid
 
@@ -149,7 +155,7 @@ export function electricCollectionOptions<T extends Row<unknown>>(
     : undefined
 
   const wrappedOnUpdate = config.onUpdate
-    ? async (params: MutationFnParams) => {
+    ? async (params: MutationFnParams<T>) => {
         const handlerResult = await config.onUpdate!(params)
         const txid = (handlerResult as { txid?: string }).txid
 
@@ -165,7 +171,7 @@ export function electricCollectionOptions<T extends Row<unknown>>(
     : undefined
 
   const wrappedOnDelete = config.onDelete
-    ? async (params: MutationFnParams) => {
+    ? async (params: MutationFnParams<T>) => {
         const handlerResult = await config.onDelete!(params)
         const txid = (handlerResult as { txid?: string }).txid
 
diff --git a/packages/db-collections/src/query.ts b/packages/db-collections/src/query.ts
index ce45353a9..a45e5388b 100644
--- a/packages/db-collections/src/query.ts
+++ b/packages/db-collections/src/query.ts
@@ -284,7 +284,7 @@ export function queryCollectionOptions<
 
   // Create wrapper handlers for direct persistence operations that handle refetching
   const wrappedOnInsert = onInsert
-    ? async (params: MutationFnParams) => {
+    ? async (params: MutationFnParams<TItem>) => {
         const handlerResult = (await onInsert(params)) ?? {}
         const shouldRefetch =
           (handlerResult as { refetch?: boolean }).refetch !== false
@@ -298,7 +298,7 @@ export function queryCollectionOptions<
     : undefined
 
   const wrappedOnUpdate = onUpdate
-    ? async (params: MutationFnParams) => {
+    ? async (params: MutationFnParams<TItem>) => {
         const handlerResult = (await onUpdate(params)) ?? {}
         const shouldRefetch =
           (handlerResult as { refetch?: boolean }).refetch !== false
@@ -312,7 +312,7 @@ export function queryCollectionOptions<
     : undefined
 
   const wrappedOnDelete = onDelete
-    ? async (params: MutationFnParams) => {
+    ? async (params: MutationFnParams<TItem>) => {
         const handlerResult = (await onDelete(params)) ?? {}
         const shouldRefetch =
           (handlerResult as { refetch?: boolean }).refetch !== false
diff --git a/packages/db-collections/tests/electric.test.ts b/packages/db-collections/tests/electric.test.ts
index 425f22fe8..a599ce0e6 100644
--- a/packages/db-collections/tests/electric.test.ts
+++ b/packages/db-collections/tests/electric.test.ts
@@ -7,6 +7,7 @@ import type {
   MutationFnParams,
   PendingMutation,
   Transaction,
+  TransactionWithMutations,
 } from "@tanstack/db"
 import type { Message, Row } from "@electric-sql/client"
 
@@ -415,8 +416,11 @@ describe(`Electric Integration`, () => {
 
     it(`should throw an error if handler doesn't return a txid`, async () => {
       // Create a mock transaction for testing
-      const mockTransaction = { id: `test-transaction` } as Transaction
-      const mockParams: MutationFnParams = { transaction: mockTransaction }
+      const mockTransaction = {
+        id: `test-transaction`,
+        mutations: [],
+      } as unknown as TransactionWithMutations<Row>
+      const mockParams: MutationFnParams<Row> = { transaction: mockTransaction }
 
       // Create a handler that doesn't return a txid
       const onInsert = vi.fn().mockResolvedValue({})
@@ -488,7 +492,7 @@ describe(`Electric Integration`, () => {
       }
 
       // Create a mutation function for the transaction
-      const mutationFn = vi.fn(async (params: MutationFnParams) => {
+      const mutationFn = vi.fn(async (params: MutationFnParams<Row>) => {
         const txid = await fakeBackend.persist(params.transaction.mutations)
 
         // Simulate server sending sync message after a delay
@@ -500,7 +504,7 @@ describe(`Electric Integration`, () => {
       })
 
       // Create direct persistence handler that returns the txid
-      const onInsert = vi.fn(async (params: MutationFnParams) => {
+      const onInsert = vi.fn(async (params: MutationFnParams<Row>) => {
         return { txid: await mutationFn(params) }
       })
 
diff --git a/packages/db/src/collection.ts b/packages/db/src/collection.ts
index 9d1f124a3..bc4f221df 100644
--- a/packages/db/src/collection.ts
+++ b/packages/db/src/collection.ts
@@ -201,7 +201,7 @@ export class CollectionImpl<T extends object = Record<string, unknown>> {
    * This is populated by createCollection
    */
   public utils: Record<string, Fn> = {}
-  public transactions: Store<SortedMap<string, TransactionType>>
+  public transactions: Store<SortedMap<string, TransactionType<any>>>
   public optimisticOperations: Derived<Array<OptimisticChangeMessage<T>>>
   public derivedState: Derived<Map<string, T>>
   public derivedArray: Derived<Array<T>>
@@ -250,7 +250,7 @@ export class CollectionImpl<T extends object = Record<string, unknown>> {
     }
 
     this.transactions = new Store(
-      new SortedMap<string, TransactionType>(
+      new SortedMap<string, TransactionType<any>>(
         (a, b) => a.createdAt.getTime() - b.createdAt.getTime()
       )
     )
@@ -269,7 +269,7 @@ export class CollectionImpl<T extends object = Record<string, unknown>> {
                 const message: OptimisticChangeMessage<T> = {
                   type: mutation.type,
                   key: mutation.key,
-                  value: mutation.modified as T,
+                  value: mutation.modified,
                   isActive,
                 }
                 if (
@@ -684,8 +684,8 @@ export class CollectionImpl<T extends object = Record<string, unknown>> {
       const mutation: PendingMutation<T> = {
         mutationId: crypto.randomUUID(),
         original: {},
-        modified: validatedData as Record<string, unknown>,
-        changes: validatedData as Record<string, unknown>,
+        modified: validatedData,
+        changes: validatedData,
         key,
         metadata: config?.metadata as unknown,
         syncMetadata: this.config.sync.getSyncMetadata?.() || {},
@@ -710,7 +710,7 @@ export class CollectionImpl<T extends object = Record<string, unknown>> {
       return ambientTransaction
     } else {
       // Create a new transaction with a mutation function that calls the onInsert handler
-      const directOpTransaction = new Transaction({
+      const directOpTransaction = new Transaction<T>({
         mutationFn: async (params) => {
           // Call the onInsert handler with the transaction
           return this.config.onInsert!(params)
@@ -906,10 +906,10 @@ export class CollectionImpl<T extends object = Record<string, unknown>> {
     // No need to check for onUpdate handler here as we've already checked at the beginning
 
     // Create a new transaction with a mutation function that calls the onUpdate handler
-    const directOpTransaction = new Transaction({
-      mutationFn: async (transaction) => {
+    const directOpTransaction = new Transaction<T>({
+      mutationFn: async (params) => {
         // Call the onUpdate handler with the transaction
-        return this.config.onUpdate!(transaction)
+        return this.config.onUpdate!(params)
       },
     })
 
@@ -944,7 +944,7 @@ export class CollectionImpl<T extends object = Record<string, unknown>> {
   delete = (
     ids: Array<string> | string,
     config?: OperationConfig
-  ): TransactionType => {
+  ): TransactionType<any> => {
     const ambientTransaction = getActiveTransaction()
 
     // If no ambient transaction exists, check for an onDelete handler early
@@ -962,9 +962,9 @@ export class CollectionImpl<T extends object = Record<string, unknown>> {
     for (const id of idsArray) {
       const mutation: PendingMutation<T> = {
         mutationId: crypto.randomUUID(),
-        original: (this.state.get(id) || {}) as Record<string, unknown>,
-        modified: (this.state.get(id) || {}) as Record<string, unknown>,
-        changes: (this.state.get(id) || {}) as Record<string, unknown>,
+        original: this.state.get(id) || {},
+        modified: this.state.get(id)!,
+        changes: this.state.get(id) || {},
         key: id,
         metadata: config?.metadata as unknown,
         syncMetadata: (this.syncedMetadata.state.get(id) || {}) as Record<
@@ -993,11 +993,11 @@ export class CollectionImpl<T extends object = Record<string, unknown>> {
     }
 
     // Create a new transaction with a mutation function that calls the onDelete handler
-    const directOpTransaction = new Transaction({
+    const directOpTransaction = new Transaction<T>({
       autoCommit: true,
-      mutationFn: async (transaction) => {
+      mutationFn: async (params) => {
         // Call the onDelete handler with the transaction
-        return this.config.onDelete!(transaction)
+        return this.config.onDelete!(params)
       },
     })
 
diff --git a/packages/db/src/transactions.ts b/packages/db/src/transactions.ts
index ccf628a67..9e3e1c600 100644
--- a/packages/db/src/transactions.ts
+++ b/packages/db/src/transactions.ts
@@ -1,6 +1,7 @@
 import { createDeferred } from "./deferred"
 import type { Deferred } from "./deferred"
 import type {
+  MutationFn,
   PendingMutation,
   TransactionConfig,
   TransactionState,
@@ -24,8 +25,8 @@ function generateUUID() {
   })
 }
 
-const transactions: Array<Transaction> = []
-let transactionStack: Array<Transaction> = []
+const transactions: Array<Transaction<any>> = []
+let transactionStack: Array<Transaction<any>> = []
 
 export function createTransaction(config: TransactionConfig): Transaction {
   if (typeof config.mutationFn === `undefined`) {
@@ -51,27 +52,27 @@ export function getActiveTransaction(): Transaction | undefined {
   }
 }
 
-function registerTransaction(tx: Transaction) {
+function registerTransaction(tx: Transaction<any>) {
   transactionStack.push(tx)
 }
 
-function unregisterTransaction(tx: Transaction) {
+function unregisterTransaction(tx: Transaction<any>) {
   transactionStack = transactionStack.filter((t) => t.id !== tx.id)
 }
 
-function removeFromPendingList(tx: Transaction) {
+function removeFromPendingList(tx: Transaction<any>) {
   const index = transactions.findIndex((t) => t.id === tx.id)
   if (index !== -1) {
     transactions.splice(index, 1)
   }
 }
 
-export class Transaction {
+export class Transaction<T extends object = Record<string, unknown>> {
   public id: string
   public state: TransactionState
-  public mutationFn
-  public mutations: Array<PendingMutation<any>>
-  public isPersisted: Deferred<Transaction>
+  public mutationFn: MutationFn<T>
+  public mutations: Array<PendingMutation<T>>
+  public isPersisted: Deferred<Transaction<T>>
   public autoCommit: boolean
   public createdAt: Date
   public metadata: Record<string, unknown>
@@ -80,12 +81,12 @@ export class Transaction {
     error: Error
   }
 
-  constructor(config: TransactionConfig) {
+  constructor(config: TransactionConfig<T>) {
     this.id = config.id!
     this.mutationFn = config.mutationFn
     this.state = `pending`
     this.mutations = []
-    this.isPersisted = createDeferred()
+    this.isPersisted = createDeferred<Transaction<T>>()
     this.autoCommit = config.autoCommit ?? true
     this.createdAt = new Date()
     this.metadata = config.metadata ?? {}
@@ -99,7 +100,7 @@ export class Transaction {
     }
   }
 
-  mutate(callback: () => void): Transaction {
+  mutate(callback: () => void): Transaction<T> {
     if (this.state !== `pending`) {
       throw `You can no longer call .mutate() as the transaction is no longer pending`
     }
@@ -134,7 +135,7 @@ export class Transaction {
     }
   }
 
-  rollback(config?: { isSecondaryRollback?: boolean }): Transaction {
+  rollback(config?: { isSecondaryRollback?: boolean }): Transaction<T> {
     const isSecondaryRollback = config?.isSecondaryRollback ?? false
     if (this.state === `completed`) {
       throw `You can no longer call .rollback() as the transaction is already completed`
@@ -173,7 +174,7 @@ export class Transaction {
     }
   }
 
-  async commit(): Promise<Transaction> {
+  async commit(): Promise<Transaction<T>> {
     if (this.state !== `pending`) {
       throw `You can no longer call .commit() as the transaction is no longer pending`
     }
@@ -189,10 +190,11 @@ export class Transaction {
     // Run mutationFn
     try {
       // At this point we know there's at least one mutation
-      // Use type assertion to tell TypeScript about this guarantee
-      const transactionWithMutations =
-        this as unknown as TransactionWithMutations
-      await this.mutationFn({ transaction: transactionWithMutations })
+      // We've already verified mutations is non-empty, so this cast is safe
+      // Use a direct type assertion instead of object spreading to preserve the original type
+      await this.mutationFn({
+        transaction: this as unknown as TransactionWithMutations<T>,
+      })
 
       this.setState(`completed`)
       this.touchCollection()
diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts
index 6cc5a188f..e90d41a66 100644
--- a/packages/db/src/types.ts
+++ b/packages/db/src/types.ts
@@ -21,9 +21,9 @@ export type UtilsRecord = Record<string, Fn>
  */
 export interface PendingMutation<T extends object = Record<string, unknown>> {
   mutationId: string
-  original: Record<string, unknown>
-  modified: Record<string, unknown>
-  changes: Record<string, unknown>
+  original: Partial<T>
+  modified: T
+  changes: Partial<T>
   key: any
   type: OperationType
   metadata: unknown
@@ -36,11 +36,18 @@ export interface PendingMutation<T extends object = Record<string, unknown>> {
 /**
  * Configuration options for creating a new transaction
  */
-export type MutationFnParams = {
-  transaction: Transaction
+export type MutationFnParams<T extends object = Record<string, unknown>> = {
+  transaction: TransactionWithMutations<T>
 }
 
-export type MutationFn = (params: MutationFnParams) => Promise<any>
+export type MutationFn<T extends object = Record<string, unknown>> = (
+  params: MutationFnParams<T>
+) => Promise<any>
+
+/**
+ * Represents a non-empty array (at least one element)
+ */
+export type NonEmptyArray<T> = [T, ...Array<T>]
 
 /**
  * Utility type for a Transaction with at least one mutation
@@ -48,16 +55,16 @@ export type MutationFn = (params: MutationFnParams) => Promise<any>
  */
 export type TransactionWithMutations<
   T extends object = Record<string, unknown>,
-> = Transaction & {
-  mutations: [PendingMutation<T>, ...Array<PendingMutation<T>>]
+> = Transaction<T> & {
+  mutations: NonEmptyArray<PendingMutation<T>>
 }
 
-export interface TransactionConfig {
+export interface TransactionConfig<T extends object = Record<string, unknown>> {
   /** Unique identifier for the transaction */
   id?: string
   /* If the transaction should autocommit after a mutate call or should commit be called explicitly */
   autoCommit?: boolean
-  mutationFn: MutationFn
+  mutationFn: MutationFn<T>
   /** Custom metadata to associate with the transaction */
   metadata?: Record<string, unknown>
 }
@@ -155,19 +162,19 @@ export interface CollectionConfig<T extends object = Record<string, unknown>> {
    * @param params Object containing transaction and mutation information
    * @returns Promise resolving to any value
    */
-  onInsert?: MutationFn
+  onInsert?: MutationFn<T>
   /**
    * Optional asynchronous handler function called before an update operation
    * @param params Object containing transaction and mutation information
    * @returns Promise resolving to any value
    */
-  onUpdate?: MutationFn
+  onUpdate?: MutationFn<T>
   /**
    * Optional asynchronous handler function called before a delete operation
    * @param params Object containing transaction and mutation information
    * @returns Promise resolving to any value
    */
-  onDelete?: MutationFn
+  onDelete?: MutationFn<T>
 }
 
 export type ChangesPayload<T extends object = Record<string, unknown>> = Array<