diff --git a/op-batcher/batcher/config.go b/op-batcher/batcher/config.go
index c6530742453..1aa99c098b9 100644
--- a/op-batcher/batcher/config.go
+++ b/op-batcher/batcher/config.go
@@ -191,6 +191,13 @@ func (c *CLIConfig) Check() error {
 	if !flags.ValidDataAvailabilityType(c.DataAvailabilityType) {
 		return fmt.Errorf("unknown data availability type: %q", c.DataAvailabilityType)
 	}
+	// The auth→batch distance (num-confirmations + inclusion delay) must fit within
+	// BatchAuthLookbackWindow, so reserve room for the batch to land or the safe head stalls.
+	const fallbackAuthInclusionReserve = 75 // blocks (~15 min at 12s L1 slots)
+	if authLookback := derive.BatchAuthLookbackWindow; authLookback < c.TxMgrConfig.NumConfirmations+fallbackAuthInclusionReserve {
+		return fmt.Errorf("NumConfirmations (%d) too high for BatchAuthLookbackWindow (%d): need %d blocks of inclusion headroom",
+			c.TxMgrConfig.NumConfirmations, authLookback, fallbackAuthInclusionReserve)
+	}
 	// Most chains' L1s still have only Cancun active, but we don't want to
 	// overcomplicate this check with a dynamic L1 query, so we just use maxBlobsPerBlock.
 	// We want to check for both, blobs and auto da-type.
diff --git a/op-batcher/batcher/config_test.go b/op-batcher/batcher/config_test.go
index 192ded7cb94..5031bbb33f5 100644
--- a/op-batcher/batcher/config_test.go
+++ b/op-batcher/batcher/config_test.go
@@ -106,6 +106,13 @@ func TestBatcherConfig(t *testing.T) {
 			override:  func(c *batcher.CLIConfig) { c.DataAvailabilityType = "foo" },
 			errString: "unknown data availability type: \"foo\"",
 		},
+		{
+			name: "num-confirmations leaves too little batch auth lookback window",
+			override: func(c *batcher.CLIConfig) {
+				c.TxMgrConfig.NumConfirmations = derive.BatchAuthLookbackWindow
+			},
+			errString: "too high for BatchAuthLookbackWindow",
+		},
 		{
 			name:      "zero TargetNumFrames",
 			override:  func(c *batcher.CLIConfig) { c.TargetNumFrames = 0 },
diff --git a/op-batcher/batcher/driver.go b/op-batcher/batcher/driver.go
index 1b0f4aa73b9..728e03a9b51 100644
--- a/op-batcher/batcher/driver.go
+++ b/op-batcher/batcher/driver.go
@@ -1049,6 +1049,9 @@ func (l *BatchSubmitter) sendTransaction(txdata txData, queue *txmgr.Queue[txRef
 
 type TxSender[T any] interface {
 	Send(id T, candidate txmgr.TxCandidate, receiptCh chan txmgr.TxReceipt[T])
+	// SendPair submits an ordered pair of transactions holding a single
+	// max-pending slot; see txmgr.Queue.SendPair.
+	SendPair(firstID T, first txmgr.TxCandidate, firstCh chan txmgr.TxReceipt[T], secondID T, second txmgr.TxCandidate, secondCh chan txmgr.TxReceipt[T])
 }
 
 // sendTx uses the txmgr queue to send the given transaction candidate after setting its
diff --git a/op-batcher/batcher/driver_test.go b/op-batcher/batcher/driver_test.go
index e5eaf518453..a1fd5e2b535 100644
--- a/op-batcher/batcher/driver_test.go
+++ b/op-batcher/batcher/driver_test.go
@@ -158,6 +158,11 @@ func (q *MockTxQueue) Send(ref txRef, candidate txmgr.TxCandidate, receiptCh cha
 	q.m.Store(ref.id.String(), candidate)
 }
 
+func (q *MockTxQueue) SendPair(firstRef txRef, first txmgr.TxCandidate, firstCh chan txmgr.TxReceipt[txRef], secondRef txRef, second txmgr.TxCandidate, secondCh chan txmgr.TxReceipt[txRef]) {
+	q.m.Store(firstRef.id.String(), first)
+	q.m.Store(secondRef.id.String(), second)
+}
+
 func (q *MockTxQueue) Load(id string) txmgr.TxCandidate {
 	c, _ := q.m.Load(id)
 	return c.(txmgr.TxCandidate)
diff --git a/op-batcher/batcher/fallback_auth.go b/op-batcher/batcher/fallback_auth.go
index 61a3de51641..419fb68d36e 100644
--- a/op-batcher/batcher/fallback_auth.go
+++ b/op-batcher/batcher/fallback_auth.go
@@ -43,6 +43,13 @@ func computeCommitment(candidate *txmgr.TxCandidate) ([32]byte, error) {
 // separate signature is needed — the L1 transaction is already signed by the TxManager's key.
 func (l *BatchSubmitter) sendTxWithFallbackAuth(txdata txData, isCancel bool, candidate *txmgr.TxCandidate, queue TxSender[txRef], receiptsCh chan txmgr.TxReceipt[txRef]) {
 	transactionReference := newTxRef(txdata, isCancel)
+	// The auth tx shares the batch txdata's identity (so a failure requeues the right frames)
+	// but is always a calldata tx. Auth failures must carry its real type: an ErrAlreadyReserved
+	// receipt labeled with the batch's blob type would make cancelBlockingTx cancel the wrong
+	// pool, leaving the reserving tx stuck.
+	authReference := transactionReference
+	authReference.isBlob = false
+	authReference.daType = DaTypeCalldata
 	l.Log.Debug("Sending fallback-authenticated L1 transaction", "txRef", transactionReference)
 
 	commitment, err := computeCommitment(candidate)
@@ -79,58 +86,96 @@ func (l *BatchSubmitter) sendTxWithFallbackAuth(txdata txData, isCancel bool, ca
 		To:     &l.RollupConfig.BatchAuthenticatorAddress,
 	}
 
-	// Private buffered channels: queue.Send forwards exactly one receipt to each, so the watcher
-	// reads exactly once per channel (even on context cancellation, the queue still emits a
-	// ctx-error receipt). These never reach handleReceipt; only the synthetic receipt below does.
-	authReceiptCh := make(chan txmgr.TxReceipt[txRef], 1)
-	batchReceiptCh := make(chan txmgr.TxReceipt[txRef], 1)
-
 	l.Log.Debug(
 		"Sending fallback authenticateBatchInfo transaction",
 		"txRef", transactionReference,
 		"commitment", hexutil.Encode(commitment[:]),
 		"address", l.RollupConfig.BatchAuthenticatorAddress.String(),
 	)
-	// Submit the auth tx then the batch tx, in order, on the publishing-loop goroutine so their
-	// nonces are assigned in submission order. Each Send blocks here when the queue is at its
-	// MaxPendingTransactions limit.
-	queue.Send(transactionReference, verifyCandidate, authReceiptCh)
-	queue.Send(transactionReference, *candidate, batchReceiptCh)
+
+	if len(candidate.Blobs) > 0 {
+		// SendPair doesn't support blobs.
+		l.sendFallbackAuthSerialized(transactionReference, authReference, verifyCandidate, candidate, queue, receiptsCh)
+		return
+	}
+
+	// Private buffered channels — Queue.SendPair forwards exactly one receipt to
+	// each, so the watcher reads exactly once per channel. These never reach
+	// handleReceipt; only the synthetic receipt from the watcher does.
+	authReceiptCh := make(chan txmgr.TxReceipt[txRef], 1)
+	batchReceiptCh := make(chan txmgr.TxReceipt[txRef], 1)
+
+	// Calldata pair: pipelined, auth leg first; see TxManager.SendPairAsync for the
+	// contract (ordering, slot accounting, cancellation of the batch leg on failure).
+	queue.SendPair(authReference, verifyCandidate, authReceiptCh, transactionReference, *candidate, batchReceiptCh)
 
 	l.authGroup.Add(1)
 	go func() {
 		defer l.authGroup.Done()
-		l.watchFallbackAuthReceipts(transactionReference, authReceiptCh, batchReceiptCh, receiptsCh)
+		l.watchFallbackAuthReceipts(transactionReference, authReference, authReceiptCh, batchReceiptCh, receiptsCh)
 	}()
 }
 
-// watchFallbackAuthReceipts collects the auth and batch receipts for a fallback-auth pair,
-// validates that the batch tx landed within the lookback window of the auth tx, and forwards a
-// single synthetic receipt keyed to the batch txData onto receiptsCh. Any failure produces an
-// error receipt so the channel manager rewinds and resubmits the frame set.
-func (l *BatchSubmitter) watchFallbackAuthReceipts(transactionReference txRef, authReceiptCh, batchReceiptCh chan txmgr.TxReceipt[txRef], receiptsCh chan txmgr.TxReceipt[txRef]) {
+// sendFallbackAuthSerialized submits an auth+batch pair serially: the auth tx is sent
+// and confirmed first, then the batch tx. It runs on the publishing-loop goroutine and
+// blocks it for the pair's full confirmation cycle, so consecutive pairs never overlap.
+func (l *BatchSubmitter) sendFallbackAuthSerialized(transactionReference, authReference txRef, verifyCandidate txmgr.TxCandidate, candidate *txmgr.TxCandidate, queue TxSender[txRef], receiptsCh chan txmgr.TxReceipt[txRef]) {
+	authReceiptCh := make(chan txmgr.TxReceipt[txRef], 1)
+	queue.Send(authReference, verifyCandidate, authReceiptCh)
 	authResult := <-authReceiptCh
-	batchResult := <-batchReceiptCh
-
-	if authResult.Err != nil {
-		l.Log.Error("Failed to send fallback authenticateBatchInfo transaction", "txRef", transactionReference, "err", authResult.Err)
+	if err := fallbackAuthResultError(authResult); err != nil {
+		l.Log.Error("Fallback authenticateBatchInfo transaction failed", "txRef", transactionReference, "err", err)
 		receiptsCh <- txmgr.TxReceipt[txRef]{
-			ID:  transactionReference,
-			Err: fmt.Errorf("failed to send fallback authenticateBatchInfo transaction: %w", authResult.Err),
+			ID:  authReference,
+			Err: err,
 		}
 		return
 	}
 
-	// txmgr returns a receipt as soon as the tx is mined, regardless of execution status. A
-	// reverted authenticateBatchInfo call emits no BatchInfoAuthenticated event, so the verifier
-	// drops the batch and the safe head stalls; report failure so the frames are re-queued. The
-	// batch inbox tx needs no such check: derivation reads its data by L1 inclusion, not by
-	// execution status.
+	// The auth tx is confirmed on L1, releasing the account's reservation in the
+	// calldata pool. Now send the blob batch tx.
+	batchReceiptCh := make(chan txmgr.TxReceipt[txRef], 1)
+	queue.Send(transactionReference, *candidate, batchReceiptCh)
+	batchResult := <-batchReceiptCh
+	l.resolveFallbackAuthPair(transactionReference, authReference, authResult, batchResult, receiptsCh)
+}
+
+// watchFallbackAuthReceipts collects the auth and batch receipts of a pipelined fallback-auth
+// pair and resolves them into a single synthetic receipt (see resolveFallbackAuthPair).
+func (l *BatchSubmitter) watchFallbackAuthReceipts(transactionReference, authReference txRef, authReceiptCh, batchReceiptCh chan txmgr.TxReceipt[txRef], receiptsCh chan txmgr.TxReceipt[txRef]) {
+	authResult := <-authReceiptCh
+	batchResult := <-batchReceiptCh
+	l.resolveFallbackAuthPair(transactionReference, authReference, authResult, batchResult, receiptsCh)
+}
+
+// fallbackAuthResultError interprets the auth leg's receipt: a send failure or a
+// mined-but-reverted auth tx both fail the pair, since a reverted authenticateBatchInfo call
+// emits no BatchInfoAuthenticated event and the verifier would drop the batch (txmgr returns a
+// receipt as soon as the tx is mined, regardless of execution status). The batch inbox tx
+// needs no such status check: derivation reads its data by L1 inclusion, not by execution
+// status.
+func fallbackAuthResultError(authResult txmgr.TxReceipt[txRef]) error {
+	if authResult.Err != nil {
+		return fmt.Errorf("failed to send fallback authenticateBatchInfo transaction: %w", authResult.Err)
+	}
 	if authResult.Receipt.Status != types.ReceiptStatusSuccessful {
-		l.Log.Error("Fallback authenticateBatchInfo transaction reverted", "txRef", transactionReference, "txHash", authResult.Receipt.TxHash)
+		return fmt.Errorf("fallback authenticateBatchInfo transaction reverted: %s", authResult.Receipt.TxHash)
+	}
+	return nil
+}
+
+// resolveFallbackAuthPair validates the receipts of an auth+batch pair — auth success, batch
+// success, and that the batch tx landed within the lookback window of the auth tx — and
+// forwards a single synthetic receipt keyed to the batch txData onto receiptsCh. Any failure
+// produces an error receipt so the channel manager rewinds and resubmits the frame set. Auth
+// failures are reported under authReference (see its construction in sendTxWithFallbackAuth
+// for why the calldata typing matters).
+func (l *BatchSubmitter) resolveFallbackAuthPair(transactionReference, authReference txRef, authResult, batchResult txmgr.TxReceipt[txRef], receiptsCh chan txmgr.TxReceipt[txRef]) {
+	if err := fallbackAuthResultError(authResult); err != nil {
+		l.Log.Error("Fallback authenticateBatchInfo transaction failed", "txRef", transactionReference, "err", err)
 		receiptsCh <- txmgr.TxReceipt[txRef]{
-			ID:  transactionReference,
-			Err: fmt.Errorf("fallback authenticateBatchInfo transaction reverted: %s", authResult.Receipt.TxHash),
+			ID:  authReference,
+			Err: err,
 		}
 		return
 	}
@@ -147,6 +192,7 @@ func (l *BatchSubmitter) watchFallbackAuthReceipts(transactionReference txRef, a
 	distance := new(big.Int).Sub(batchResult.Receipt.BlockNumber, authResult.Receipt.BlockNumber)
 	lookbackWindow := new(big.Int).SetUint64(derive.BatchAuthLookbackWindow)
 	if distance.Sign() < 0 || distance.Cmp(lookbackWindow) > 0 {
+		l.Metr.RecordFallbackAuthWindowExceeded()
 		l.Log.Error("authenticateBatchInfo transaction too far from batch inbox transaction", "txRef", transactionReference, "distance", distance)
 		receiptsCh <- txmgr.TxReceipt[txRef]{
 			ID:  transactionReference,
diff --git a/op-batcher/batcher/fallback_auth_test.go b/op-batcher/batcher/fallback_auth_test.go
index 0822f8a091f..f0b248dc21a 100644
--- a/op-batcher/batcher/fallback_auth_test.go
+++ b/op-batcher/batcher/fallback_auth_test.go
@@ -10,6 +10,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/stretchr/testify/require"
 
+	"github.com/ethereum-optimism/optimism/op-batcher/metrics"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
@@ -19,31 +20,54 @@ import (
 
 var errSendFailed = errors.New("send failed")
 
-// recordedSend captures a single queue.Send invocation.
+// recordedSend captures a single queued transaction, whether it arrived via
+// Send or as a leg of SendPair.
 type recordedSend struct {
 	candidate txmgr.TxCandidate
 	receiptCh chan txmgr.TxReceipt[txRef]
+	viaPair   bool
 }
 
-// fakeTxSender records Send calls in order and immediately delivers a canned
-// response (by index) on the receipt channel, mimicking the txmgr Queue, which
-// forwards exactly one receipt per Send.
+// fakeTxSender records queued transactions in order and immediately delivers a
+// canned response (by index) on the receipt channel, mimicking the txmgr
+// Queue, which forwards exactly one receipt per queued transaction. SendPair
+// records its two legs as two consecutive sends consuming two consecutive
+// canned responses.
 type fakeTxSender struct {
 	sends     []recordedSend
+	pairCalls int
 	responses []txmgr.TxReceipt[txRef]
 }
 
-func (f *fakeTxSender) Send(id txRef, candidate txmgr.TxCandidate, receiptCh chan txmgr.TxReceipt[txRef]) {
+func (f *fakeTxSender) deliver(id txRef, candidate txmgr.TxCandidate, receiptCh chan txmgr.TxReceipt[txRef], viaPair bool) {
 	idx := len(f.sends)
-	f.sends = append(f.sends, recordedSend{candidate: candidate, receiptCh: receiptCh})
+	f.sends = append(f.sends, recordedSend{candidate: candidate, receiptCh: receiptCh, viaPair: viaPair})
 	resp := f.responses[idx]
 	resp.ID = id
 	receiptCh <- resp
 }
 
+func (f *fakeTxSender) Send(id txRef, candidate txmgr.TxCandidate, receiptCh chan txmgr.TxReceipt[txRef]) {
+	f.deliver(id, candidate, receiptCh, false)
+}
+
+func (f *fakeTxSender) SendPair(firstID txRef, first txmgr.TxCandidate, firstCh chan txmgr.TxReceipt[txRef], secondID txRef, second txmgr.TxCandidate, secondCh chan txmgr.TxReceipt[txRef]) {
+	f.pairCalls++
+	f.deliver(firstID, first, firstCh, true)
+	f.deliver(secondID, second, secondCh, true)
+}
+
+type windowExceededSpy struct {
+	metrics.Metricer
+	count int
+}
+
+func (s *windowExceededSpy) RecordFallbackAuthWindowExceeded() { s.count++ }
+
 func newFallbackAuthSubmitter(t *testing.T) *BatchSubmitter {
 	l := &BatchSubmitter{}
 	l.Log = testlog.Logger(t, log.LevelDebug)
+	l.Metr = metrics.NoopMetrics
 	l.RollupConfig = &rollup.Config{
 		BatchAuthenticatorAddress: common.HexToAddress("0x00000000000000000000000000000000000000aa"),
 	}
@@ -54,6 +78,12 @@ func testFallbackTxData(t *testing.T) txData {
 	return singleFrameTxData(frameData{data: []byte("frame-data")})
 }
 
+// testBlobCandidate returns a tx candidate carrying one (zero) blob, which
+// routes sendTxWithFallbackAuth onto the serialized blob path.
+func testBlobCandidate() *txmgr.TxCandidate {
+	return &txmgr.TxCandidate{Blobs: []*eth.Blob{{}}}
+}
+
 func receiptWithBlock(num int64) *types.Receipt {
 	return &types.Receipt{BlockNumber: big.NewInt(num), Status: types.ReceiptStatusSuccessful}
 }
@@ -62,10 +92,11 @@ func revertedReceiptWithBlock(num int64) *types.Receipt {
 	return &types.Receipt{BlockNumber: big.NewInt(num), Status: types.ReceiptStatusFailed}
 }
 
-// TestFallbackAuth_OrderingAndSuccess verifies the auth tx is submitted before
-// the batch tx (so it takes the lower nonce and lands first, as Espresso
-// requires) and that a single success receipt for the batch txData is emitted
-// when both txs land within the lookback window.
+// TestFallbackAuth_OrderingAndSuccess verifies a calldata batch is submitted
+// as a single pipelined pair — the auth tx first (so it takes the lower nonce
+// and lands first, as Espresso requires), the batch tx second — and that a
+// single success receipt for the batch txData is emitted when both txs land
+// within the lookback window.
 func TestFallbackAuth_OrderingAndSuccess(t *testing.T) {
 	l := newFallbackAuthSubmitter(t)
 	txdata := testFallbackTxData(t)
@@ -80,14 +111,16 @@ func TestFallbackAuth_OrderingAndSuccess(t *testing.T) {
 	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
 
 	l.sendTxWithFallbackAuth(txdata, false, candidate, queue, receiptsCh)
-	l.authGroup.Wait()
 
 	require.Len(t, queue.sends, 2)
-	// First send must target the BatchAuthenticator (the auth tx), giving it the
+	require.Equal(t, 1, queue.pairCalls, "calldata pair must be submitted via SendPair")
+	require.True(t, queue.sends[0].viaPair)
+	require.True(t, queue.sends[1].viaPair)
+	// First leg must target the BatchAuthenticator (the auth tx), giving it the
 	// lower, earlier-mined nonce.
 	require.NotNil(t, queue.sends[0].candidate.To)
 	require.Equal(t, l.RollupConfig.BatchAuthenticatorAddress, *queue.sends[0].candidate.To)
-	// Second send is the batch tx itself.
+	// Second leg is the batch tx itself.
 	require.Equal(t, candidate.TxData, queue.sends[1].candidate.TxData)
 
 	got := <-receiptsCh
@@ -96,6 +129,11 @@ func TestFallbackAuth_OrderingAndSuccess(t *testing.T) {
 	require.Equal(t, txdata.ID().String(), got.ID.id.String())
 }
 
+// TestFallbackAuth_AuthFailureRetried verifies that an auth-leg failure of a
+// pipelined pair produces an error receipt keyed to the batch txData so the
+// frames are re-queued. Both legs enter the queue back-to-back (pipelined);
+// the tx manager cancels the batch leg when the auth leg fails, surfacing as
+// an ErrPairLegCancelled response on the batch channel.
 func TestFallbackAuth_AuthFailureRetried(t *testing.T) {
 	l := newFallbackAuthSubmitter(t)
 	txdata := testFallbackTxData(t)
@@ -103,20 +141,50 @@ func TestFallbackAuth_AuthFailureRetried(t *testing.T) {
 
 	queue := &fakeTxSender{
 		responses: []txmgr.TxReceipt[txRef]{
-			{Err: errSendFailed},             // auth fails
-			{Receipt: receiptWithBlock(101)}, // batch lands anyway
+			{Err: errSendFailed},             // auth leg fails
+			{Err: txmgr.ErrPairLegCancelled}, // batch leg cancelled by the pair
+		},
+	}
+	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
+
+	l.sendTxWithFallbackAuth(txdata, false, candidate, queue, receiptsCh)
+
+	got := <-receiptsCh
+	require.Error(t, got.Err)
+	require.Equal(t, txdata.ID().String(), got.ID.id.String())
+	require.Len(t, queue.sends, 2)
+	require.Equal(t, 1, queue.pairCalls)
+}
+
+// TestFallbackAuth_AuthRevertedRetried verifies that an authenticateBatchInfo tx
+// that mines but reverts (no event emitted for the verifier) produces an error
+// receipt so the frames are re-queued, rather than being confirmed as success.
+// The tx manager treats a reverted auth leg as a pair failure and cancels the
+// batch leg.
+func TestFallbackAuth_AuthRevertedRetried(t *testing.T) {
+	l := newFallbackAuthSubmitter(t)
+	txdata := testFallbackTxData(t)
+	candidate := &txmgr.TxCandidate{TxData: []byte("batch-calldata")}
+
+	queue := &fakeTxSender{
+		responses: []txmgr.TxReceipt[txRef]{
+			{Receipt: revertedReceiptWithBlock(100)}, // auth mined but reverted
+			{Err: txmgr.ErrPairLegCancelled},         // batch leg cancelled by the pair
 		},
 	}
 	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
 
 	l.sendTxWithFallbackAuth(txdata, false, candidate, queue, receiptsCh)
-	l.authGroup.Wait()
 
 	got := <-receiptsCh
 	require.Error(t, got.Err)
 	require.Equal(t, txdata.ID().String(), got.ID.id.String())
+	require.Len(t, queue.sends, 2)
+	require.Equal(t, 1, queue.pairCalls)
 }
 
+// TestFallbackAuth_BatchFailureRetried verifies a batch-leg failure produces an
+// error receipt keyed to the batch txData.
 func TestFallbackAuth_BatchFailureRetried(t *testing.T) {
 	l := newFallbackAuthSubmitter(t)
 	txdata := testFallbackTxData(t)
@@ -131,35 +199,146 @@ func TestFallbackAuth_BatchFailureRetried(t *testing.T) {
 	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
 
 	l.sendTxWithFallbackAuth(txdata, false, candidate, queue, receiptsCh)
-	l.authGroup.Wait()
 
 	got := <-receiptsCh
 	require.Error(t, got.Err)
 	require.Equal(t, txdata.ID().String(), got.ID.id.String())
+	require.Equal(t, 1, queue.pairCalls)
 }
 
-// TestFallbackAuth_AuthRevertedRetried verifies that an authenticateBatchInfo tx
-// that mines but reverts (no event emitted for the verifier) produces an error
-// receipt so the frames are re-queued, rather than being confirmed as success.
-func TestFallbackAuth_AuthRevertedRetried(t *testing.T) {
+// TestFallbackAuth_BlobSerialized verifies a blob batch takes the serialized
+// path: the auth tx is sent alone (not as a pair) and the blob batch tx is
+// only sent after the auth tx confirms, because geth's cross-pool account
+// reservation cannot hold a calldata auth tx and a blob batch tx at once.
+func TestFallbackAuth_BlobSerialized(t *testing.T) {
 	l := newFallbackAuthSubmitter(t)
 	txdata := testFallbackTxData(t)
-	candidate := &txmgr.TxCandidate{TxData: []byte("batch-calldata")}
+	txdata.daType = DaTypeBlob
+	candidate := testBlobCandidate()
+
+	queue := &fakeTxSender{
+		responses: []txmgr.TxReceipt[txRef]{
+			{Receipt: receiptWithBlock(100)}, // auth
+			{Receipt: receiptWithBlock(101)}, // batch, sent only after auth confirmed
+		},
+	}
+	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
+
+	l.sendTxWithFallbackAuth(txdata, false, candidate, queue, receiptsCh)
+
+	require.Len(t, queue.sends, 2)
+	require.Equal(t, 0, queue.pairCalls, "blob pair must not be pipelined")
+	require.False(t, queue.sends[0].viaPair)
+	require.False(t, queue.sends[1].viaPair)
+	require.Equal(t, l.RollupConfig.BatchAuthenticatorAddress, *queue.sends[0].candidate.To)
+
+	got := <-receiptsCh
+	require.NoError(t, got.Err)
+	require.Equal(t, txdata.ID().String(), got.ID.id.String())
+}
+
+// TestFallbackAuth_BlobAuthFailureGatesBatch verifies the serialized blob path
+// never sends the batch tx when the auth tx fails: the batch would be crafted
+// at the next nonce while the failed auth send resets the txmgr nonce, leaving
+// a nonce gap (and an unauthenticated blob batch) behind.
+func TestFallbackAuth_BlobAuthFailureGatesBatch(t *testing.T) {
+	l := newFallbackAuthSubmitter(t)
+	txdata := testFallbackTxData(t)
+	txdata.daType = DaTypeBlob
+	candidate := testBlobCandidate()
+
+	queue := &fakeTxSender{
+		responses: []txmgr.TxReceipt[txRef]{
+			{Err: errSendFailed}, // auth fails. No batch response, it must never be sent
+		},
+	}
+	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
+
+	l.sendTxWithFallbackAuth(txdata, false, candidate, queue, receiptsCh)
+
+	got := <-receiptsCh
+	require.Error(t, got.Err)
+	require.Equal(t, txdata.ID().String(), got.ID.id.String())
+	require.Len(t, queue.sends, 1)
+}
+
+// TestFallbackAuth_BlobAuthRevertGatesBatch is the revert variant: a mined but
+// reverted auth emits no BatchInfoAuthenticated event, so the blob batch would
+// be unverifiable and must not be submitted at all.
+func TestFallbackAuth_BlobAuthRevertGatesBatch(t *testing.T) {
+	l := newFallbackAuthSubmitter(t)
+	txdata := testFallbackTxData(t)
+	txdata.daType = DaTypeBlob
+	candidate := testBlobCandidate()
 
 	queue := &fakeTxSender{
 		responses: []txmgr.TxReceipt[txRef]{
 			{Receipt: revertedReceiptWithBlock(100)}, // auth mined but reverted
-			{Receipt: receiptWithBlock(101)},         // batch lands
 		},
 	}
 	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
 
 	l.sendTxWithFallbackAuth(txdata, false, candidate, queue, receiptsCh)
-	l.authGroup.Wait()
 
 	got := <-receiptsCh
 	require.Error(t, got.Err)
 	require.Equal(t, txdata.ID().String(), got.ID.id.String())
+	require.Len(t, queue.sends, 1)
+}
+
+// TestFallbackAuth_AuthFailureTxRefType verifies that an auth-tx failure is
+// reported under a calldata-typed txRef even when the batch txdata is blob.
+// The auth tx is always calldata; if its ErrAlreadyReserved failure were
+// labeled with the batch's blob type, cancelBlockingTx would send a calldata
+// cancel against a blobpool reservation, which is rejected the same way,
+// looping forever without ever displacing the stuck blob tx.
+func TestFallbackAuth_AuthFailureTxRefType(t *testing.T) {
+	l := newFallbackAuthSubmitter(t)
+	txdata := testFallbackTxData(t)
+	txdata.daType = DaTypeBlob
+	candidate := testBlobCandidate()
+
+	queue := &fakeTxSender{
+		responses: []txmgr.TxReceipt[txRef]{
+			{Err: errSendFailed}, // auth fails. No batch response, it must never be sent
+		},
+	}
+	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
+
+	l.sendTxWithFallbackAuth(txdata, false, candidate, queue, receiptsCh)
+
+	got := <-receiptsCh
+	require.Error(t, got.Err)
+	require.Equal(t, txdata.ID().String(), got.ID.id.String())
+	require.Len(t, queue.sends, 1)
+	require.False(t, got.ID.isBlob)
+	require.Equal(t, DaTypeCalldata, got.ID.daType)
+}
+
+// TestFallbackAuth_BatchFailureTxRefType verifies the converse: a batch-tx
+// failure keeps the batch txdata's own type on the forwarded receipt.
+func TestFallbackAuth_BatchFailureTxRefType(t *testing.T) {
+	l := newFallbackAuthSubmitter(t)
+	txdata := testFallbackTxData(t)
+	txdata.daType = DaTypeBlob
+	candidate := testBlobCandidate()
+
+	queue := &fakeTxSender{
+		responses: []txmgr.TxReceipt[txRef]{
+			{Receipt: receiptWithBlock(100)}, // auth lands
+			{Err: errSendFailed},             // batch fails
+		},
+	}
+	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
+
+	l.sendTxWithFallbackAuth(txdata, false, candidate, queue, receiptsCh)
+
+	got := <-receiptsCh
+	require.Error(t, got.Err)
+	require.Equal(t, txdata.ID().String(), got.ID.id.String())
+	require.Len(t, queue.sends, 2)
+	require.True(t, got.ID.isBlob)
+	require.Equal(t, DaTypeBlob, got.ID.daType)
 }
 
 // TestFallbackAuth_WindowViolationRetried verifies that a batch tx landing
@@ -167,6 +346,8 @@ func TestFallbackAuth_AuthRevertedRetried(t *testing.T) {
 // channel manager rewinds and resubmits), rather than being confirmed.
 func TestFallbackAuth_WindowViolationRetried(t *testing.T) {
 	l := newFallbackAuthSubmitter(t)
+	metr := &windowExceededSpy{Metricer: metrics.NoopMetrics}
+	l.Metr = metr
 	txdata := testFallbackTxData(t)
 	candidate := &txmgr.TxCandidate{TxData: []byte("batch-calldata")}
 
@@ -180,11 +361,11 @@ func TestFallbackAuth_WindowViolationRetried(t *testing.T) {
 	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
 
 	l.sendTxWithFallbackAuth(txdata, false, candidate, queue, receiptsCh)
-	l.authGroup.Wait()
 
 	got := <-receiptsCh
 	require.Error(t, got.Err)
 	require.Equal(t, txdata.ID().String(), got.ID.id.String())
+	require.Equal(t, 1, metr.count, "window violation should record the fallback_auth_window_exceeded metric")
 }
 
 // TestFallbackAuth_WindowBoundaryAccepted pins the inclusive bound of the lookback
@@ -207,7 +388,6 @@ func TestFallbackAuth_WindowBoundaryAccepted(t *testing.T) {
 	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
 
 	l.sendTxWithFallbackAuth(txdata, false, candidate, queue, receiptsCh)
-	l.authGroup.Wait()
 
 	got := <-receiptsCh
 	require.NoError(t, got.Err)
diff --git a/op-batcher/metrics/metrics.go b/op-batcher/metrics/metrics.go
index 93c8efd2808..84057122a0b 100644
--- a/op-batcher/metrics/metrics.go
+++ b/op-batcher/metrics/metrics.go
@@ -69,6 +69,8 @@ type Metricer interface {
 	RecordBatchDataSizeBytes(daType string, size int)
 	RecordFailoverToEthDA()
 
+	RecordFallbackAuthWindowExceeded()
+
 	Document() []opmetrics.DocumentedMetric
 
 	PendingDABytes() float64
@@ -110,9 +112,10 @@ type Metrics struct {
 	channelOutputBytesTotal prometheus.Counter
 	channelQueueLength      prometheus.Gauge
 
-	batchSentDATypeTotal          prometheus.CounterVec
-	batchStoredDataSizeBytesTotal prometheus.CounterVec
-	altDaFailoverTotal            prometheus.Counter
+	batchSentDATypeTotal            prometheus.CounterVec
+	batchStoredDataSizeBytesTotal   prometheus.CounterVec
+	altDaFailoverTotal              prometheus.Counter
+	fallbackAuthWindowExceededTotal prometheus.Counter
 
 	batcherTxEvs opmetrics.EventVec
 
@@ -255,6 +258,11 @@ func NewMetrics(procName string) *Metrics {
 			Name:      "alt_da_failover_total",
 			Help:      "Total number of batches that could not be stored in AltDA and were sent to L1 instead",
 		}),
+		fallbackAuthWindowExceededTotal: factory.NewCounter(prometheus.CounterOpts{
+			Namespace: ns,
+			Name:      "fallback_auth_window_exceeded_total",
+			Help:      "Total number of fallback-auth submissions dropped because the batch tx landed beyond BatchAuthLookbackWindow blocks after its auth tx",
+		}),
 		blobUsedBytes: factory.NewHistogram(prometheus.HistogramOpts{
 			Namespace: ns,
 			Name:      "blob_used_bytes",
@@ -483,6 +491,10 @@ func (m *Metrics) RecordFailoverToEthDA() {
 	m.altDaFailoverTotal.Inc()
 }
 
+func (m *Metrics) RecordFallbackAuthWindowExceeded() {
+	m.fallbackAuthWindowExceededTotal.Inc()
+}
+
 func (m *Metrics) RecordChannelQueueLength(len int) {
 	m.channelQueueLength.Set(float64(len))
 }
diff --git a/op-batcher/metrics/noop.go b/op-batcher/metrics/noop.go
index 31002349fe2..b44bb8950f1 100644
--- a/op-batcher/metrics/noop.go
+++ b/op-batcher/metrics/noop.go
@@ -64,6 +64,7 @@ func (*noopMetrics) RecordBlobUsedBytes(int) {}
 func (*noopMetrics) RecordBatchDaType(string)             {}
 func (*noopMetrics) RecordBatchDataSizeBytes(string, int) {}
 func (*noopMetrics) RecordFailoverToEthDA()               {}
+func (*noopMetrics) RecordFallbackAuthWindowExceeded()    {}
 
 func (*noopMetrics) StartBalanceMetrics(log.Logger, *ethclient.Client, common.Address) io.Closer {
 	return nil
diff --git a/op-challenger/sender/sender_test.go b/op-challenger/sender/sender_test.go
index 4852ef3e664..31029cf46a5 100644
--- a/op-challenger/sender/sender_test.go
+++ b/op-challenger/sender/sender_test.go
@@ -141,6 +141,10 @@ func (s *stubTxMgr) SendAsync(ctx context.Context, candidate txmgr.TxCandidate,
 	}()
 }
 
+func (s *stubTxMgr) SendPairAsync(ctx context.Context, first txmgr.TxCandidate, second txmgr.TxCandidate, firstCh chan txmgr.SendResponse, secondCh chan txmgr.SendResponse) {
+	panic("unimplemented")
+}
+
 func (s *stubTxMgr) recordTx(candidate txmgr.TxCandidate) chan *types.Receipt {
 	s.m.Lock()
 	defer s.m.Unlock()
diff --git a/op-e2e/actions/helpers/l2_proposer.go b/op-e2e/actions/helpers/l2_proposer.go
index 2df068d077d..50e002ba852 100644
--- a/op-e2e/actions/helpers/l2_proposer.go
+++ b/op-e2e/actions/helpers/l2_proposer.go
@@ -81,6 +81,10 @@ func (f fakeTxMgr) SendAsync(ctx context.Context, candidate txmgr.TxCandidate, c
 	panic("unimplemented")
 }
 
+func (f fakeTxMgr) SendPairAsync(ctx context.Context, first txmgr.TxCandidate, second txmgr.TxCandidate, firstCh chan txmgr.SendResponse, secondCh chan txmgr.SendResponse) {
+	panic("unimplemented")
+}
+
 func (f fakeTxMgr) Close() {
 }
 
diff --git a/op-service/testutils/fake_txmgr.go b/op-service/testutils/fake_txmgr.go
index 3f8d309f09d..a049b55ede9 100644
--- a/op-service/testutils/fake_txmgr.go
+++ b/op-service/testutils/fake_txmgr.go
@@ -60,6 +60,10 @@ func (f *FakeTxMgr) SendAsync(ctx context.Context, candidate txmgr.TxCandidate,
 	}
 	ch <- sendResponse
 }
+func (f *FakeTxMgr) SendPairAsync(ctx context.Context, first txmgr.TxCandidate, second txmgr.TxCandidate, firstCh chan txmgr.SendResponse, secondCh chan txmgr.SendResponse) {
+	f.SendAsync(ctx, first, firstCh)
+	f.SendAsync(ctx, second, secondCh)
+}
 func (f *FakeTxMgr) ChainID() eth.ChainID {
 	return f.chainId
 }
diff --git a/op-service/txmgr/leg.go b/op-service/txmgr/leg.go
new file mode 100644
index 00000000000..63a70af2353
--- /dev/null
+++ b/op-service/txmgr/leg.go
@@ -0,0 +1,36 @@
+package txmgr
+
+import (
+	"context"
+
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+// leg holds everything belonging to one leg of a SendPairAsync pair: the
+// signed transaction, the context its submission loop runs on (cancelled by
+// the pair when it resolves), and the channel its final response is delivered
+// on.
+type leg struct {
+	sendContext context.Context
+	cancelFunc  context.CancelFunc
+	tx          *types.Transaction
+	responseCh  chan SendResponse
+}
+
+// legResult is the outcome of one leg's submission loop.
+type legResult struct {
+	receipt *types.Receipt
+	err     error
+}
+
+// send runs the leg's transaction through the manager's full submission loop
+// on the leg's own context.
+func (l *leg) send(m *SimpleTxManager) legResult {
+	receipt, err := m.sendTx(l.sendContext, l.tx)
+	return legResult{receipt, err}
+}
+
+// respond delivers the leg's final response on its response channel.
+func (l *leg) respond(res legResult) {
+	l.responseCh <- SendResponse{Receipt: res.receipt, Nonce: l.tx.Nonce(), Err: res.err}
+}
diff --git a/op-service/txmgr/mocks/TxManager.go b/op-service/txmgr/mocks/TxManager.go
index db3acedf199..0f282b9abb9 100644
--- a/op-service/txmgr/mocks/TxManager.go
+++ b/op-service/txmgr/mocks/TxManager.go
@@ -168,6 +168,11 @@ func (_m *TxManager) SendAsync(ctx context.Context, candidate txmgr.TxCandidate,
 	_m.Called(ctx, candidate, ch)
 }
 
+// SendPairAsync provides a mock function with given fields: ctx, first, second, firstCh, secondCh
+func (_m *TxManager) SendPairAsync(ctx context.Context, first txmgr.TxCandidate, second txmgr.TxCandidate, firstCh chan txmgr.SendResponse, secondCh chan txmgr.SendResponse) {
+	_m.Called(ctx, first, second, firstCh, secondCh)
+}
+
 // SuggestGasPriceCaps provides a mock function with given fields: ctx
 func (_m *TxManager) SuggestGasPriceCaps(ctx context.Context) (*big.Int, *big.Int, *big.Int, *big.Int, error) {
 	ret := _m.Called(ctx)
diff --git a/op-service/txmgr/pair.go b/op-service/txmgr/pair.go
new file mode 100644
index 00000000000..c166e0de78e
--- /dev/null
+++ b/op-service/txmgr/pair.go
@@ -0,0 +1,246 @@
+package txmgr
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/params"
+
+	"github.com/ethereum-optimism/optimism/op-service/retry"
+)
+
+// ErrPairLegCancelled is delivered as the second leg's response error when the
+// first leg of a SendPairAsync pair fails permanently and the second leg is
+// cancelled as a result.
+var ErrPairLegCancelled = errors.New("paired transaction cancelled: first leg failed")
+
+// errPairBlobsUnsupported is returned when a pair candidate carries blobs;
+// see the blob paragraph in SendPairAsync's doc for why.
+var errPairBlobsUnsupported = errors.New("SendPairAsync does not support blob candidates")
+
+func getContext(ctx context.Context, txTimeout time.Duration) (context.Context, context.CancelFunc) {
+	if txTimeout == 0 {
+		return context.WithCancel(ctx)
+	}
+	return context.WithTimeout(ctx, txTimeout)
+}
+
+// pairSend drives one SendPairAsync submission from broadcast through repair
+// and response delivery.
+type pairSend struct {
+	txManager *SimpleTxManager
+	parentCtx context.Context
+	leg1      leg
+	leg2      leg
+}
+
+// sendPair drives the pair to resolution on its own goroutine: it broadcasts both
+// legs concurrently, cancels the second leg if the first fails permanently,
+// repairs any nonce the pair would otherwise leave unmined, and delivers
+// exactly one response per channel once the pair's on-chain state is clean.
+func (p *pairSend) sendPair() {
+	m := p.txManager
+	defer func() { m.metr.RecordPendingTx(m.pending.Add(-2)) }()
+	defer p.leg1.cancelFunc()
+	defer p.leg2.cancelFunc()
+
+	leg2Res := make(chan legResult, 1)
+	go func() {
+		leg2Res <- p.leg2.send(m)
+	}()
+
+	legResult1 := p.leg1.send(m)
+
+	// A mined-but-reverted first leg also fails the pair, not just a failed
+	// send: the second leg depends on the first leg's successful execution.
+	leg1Failed := legResult1.err != nil || (legResult1.receipt != nil && legResult1.receipt.Status != types.ReceiptStatusSuccessful)
+	if leg1Failed {
+		m.l.Warn("pair first leg failed; cancelling second leg",
+			"firstNonce", p.leg1.tx.Nonce(), "secondNonce", p.leg2.tx.Nonce(),
+			"err", legResult1.err, "reverted", legResult1.err == nil)
+		p.leg2.cancelFunc()
+	}
+
+	legResult2 := <-leg2Res
+
+	p.repair(legResult1.err, legResult2.err)
+
+	// A second leg that errored after the first leg failed did not fail on
+	// its own — it was cancelled because the pair failed; report it as such.
+	if leg1Failed && legResult2.err != nil {
+		legResult2.err = fmt.Errorf("%w: %w", ErrPairLegCancelled, legResult2.err)
+	}
+	p.leg1.respond(legResult1)
+	p.leg2.respond(legResult2)
+}
+
+// repair consumes any nonce the pair would otherwise leave unmined: a leg
+// that resolved with an error never mined, so its nonce gets a cancellation
+// no-op (see cancelPairNonce)
+func (p *pairSend) repair(err1 error, err2 error) {
+	// Nothing to repair when both legs mined
+	if err1 == nil && err2 == nil {
+		return
+	}
+
+	m := p.txManager
+	if p.parentCtx.Err() != nil {
+		m.l.Info("pair send context cancelled; leaving in-flight legs to resolve in the pool",
+			"firstNonce", p.leg1.tx.Nonce(), "secondNonce", p.leg2.tx.Nonce())
+		return
+	}
+
+	repairCtx, cancel := getContext(context.WithoutCancel(p.parentCtx), m.cfg.TxSendTimeout)
+	defer cancel()
+
+	var repairs sync.WaitGroup
+	if err1 != nil {
+		repairs.Add(1)
+		go func() {
+			defer repairs.Done()
+			m.cancelPairNonce(repairCtx, p.leg1.tx)
+		}()
+	}
+	if err2 != nil {
+		repairs.Add(1)
+		go func() {
+			defer repairs.Done()
+			m.cancelPairNonce(repairCtx, p.leg2.tx)
+		}()
+	}
+	repairs.Wait()
+}
+
+// SendPairAsync implements TxManager.SendPairAsync — see the interface doc
+// for the pair contract. It crafts both legs synchronously and hands them to
+// pairSend.sendPair; cancellation and nonce repair live in repair and
+// cancelPairNonce.
+func (m *SimpleTxManager) SendPairAsync(
+	ctx context.Context,
+	firstCandidate TxCandidate,
+	secondCandidate TxCandidate,
+	firstRespCh chan SendResponse,
+	secondRespCh chan SendResponse,
+) {
+	if cap(firstRespCh) == 0 || cap(secondRespCh) == 0 {
+		panic("SendPairAsync: channels must be buffered")
+	}
+
+	respondErr := func(err1 error, err2 error) {
+		firstRespCh <- SendResponse{Err: err1}
+		secondRespCh <- SendResponse{Err: err2}
+	}
+
+	if len(firstCandidate.Blobs) > 0 || len(secondCandidate.Blobs) > 0 {
+		respondErr(errPairBlobsUnsupported, errPairBlobsUnsupported)
+		return
+	}
+
+	// refuse new requests if the tx manager is closed
+	if m.closed.Load() {
+		respondErr(ErrClosed, ErrClosed)
+		return
+	}
+
+	prepareCtx, cancel := getContext(ctx, m.cfg.TxSendTimeout)
+	tx1, err := m.prepare(prepareCtx, firstCandidate)
+	if err != nil {
+		m.resetNonce()
+		cancel()
+		respondErr(err, fmt.Errorf("%w: first leg preparation failed: %w", ErrPairLegCancelled, err))
+		return
+	}
+
+	prepareCtx, cancel = getContext(ctx, m.cfg.TxSendTimeout)
+	tx2, err := m.prepare(prepareCtx, secondCandidate)
+	if err != nil {
+		m.resetNonce()
+		cancel()
+		respondErr(fmt.Errorf("pair aborted before broadcast: second leg preparation failed: %w", err), err)
+		return
+	}
+
+	leg1 := leg{responseCh: firstRespCh, tx: tx1}
+	leg2 := leg{responseCh: secondRespCh, tx: tx2}
+	leg1.sendContext, leg1.cancelFunc = getContext(ctx, m.cfg.TxSendTimeout)
+	leg2.sendContext, leg2.cancelFunc = getContext(ctx, m.cfg.TxSendTimeout)
+
+	m.metr.RecordPendingTx(m.pending.Add(2))
+
+	p := &pairSend{
+		txManager: m,
+		parentCtx: ctx,
+		leg1:      leg1,
+		leg2:      leg2,
+	}
+	go p.sendPair()
+}
+
+// cancelPairNonce consumes target's nonce on chain by publishing a fee-bumped
+// no-op (zero-value self-send) at that exact nonce, and waits for the nonce
+// to be consumed — by the no-op, or by target itself if it wins the race
+// (both outcomes end with the nonce deterministically spent). This is what
+// lets a failed pair guarantee it leaves no gap and no live orphan behind
+// without a global nonce reset.
+//
+// Cancellation is a replacement: the no-op's fees start bumped above target's
+// original fees (satisfying the pool's replacement rules), and sendTx's
+// underpriced handling keeps bumping if a higher-fee version of target is
+// still floating. If the fee limit prevents out-pricing target, target itself
+// is high-fee enough to mine, which also consumes the nonce.
+//
+// On failure (e.g. RPC down for the whole attempt) it falls back to
+// resetNonce, restoring the legacy behavior where the next send re-queries
+// the chain nonce and refills the gap.
+func (m *SimpleTxManager) cancelPairNonce(ctx context.Context, target *types.Transaction) {
+	l := m.l.New("nonce", target.Nonce(), "replaces", target.Hash())
+	noopTx, err := retry.Do(ctx, 5, retry.Fixed(2*time.Second), func() (*types.Transaction, error) {
+		return m.craftPairCancelTx(ctx, target)
+	})
+	if err != nil {
+		l.Error("failed to craft pair cancellation tx; falling back to nonce reset", "err", err)
+		m.resetNonce()
+		return
+	}
+
+	if _, err := m.sendTx(ctx, noopTx); err != nil {
+		if errors.Is(err, core.ErrNonceTooLow) {
+			// The nonce was consumed by another transaction — target itself,
+			// or an earlier cancellation. Either way it is spent, which is
+			// all the pair needs.
+			l.Info("pair nonce consumed by another transaction before cancellation")
+			return
+		}
+		l.Error("failed to cancel pair nonce; falling back to nonce reset", "err", err)
+		m.resetNonce()
+		return
+	}
+	l.Info("pair nonce cancelled", "cancelTx", noopTx.Hash())
+}
+
+// craftPairCancelTx builds and signs the no-op replacement
+func (m *SimpleTxManager) craftPairCancelTx(ctx context.Context, target *types.Transaction) (*types.Transaction, error) {
+	tip, baseFee, _, _, err := m.SuggestGasPriceCaps(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get gas price info: %w", err)
+	}
+	bumpedTip, bumpedFee := updateFees(target.GasTipCap(), target.GasFeeCap(), tip, baseFee, false, m.l)
+
+	to := m.cfg.From
+	msg := &types.DynamicFeeTx{
+		ChainID:   m.chainID,
+		Nonce:     target.Nonce(),
+		To:        &to,
+		GasTipCap: bumpedTip,
+		GasFeeCap: bumpedFee,
+		Gas:       params.TxGas,
+	}
+	signerCtx, cancel := context.WithTimeout(ctx, m.cfg.NetworkTimeout)
+	defer cancel()
+	return m.cfg.Signer(signerCtx, m.cfg.From, types.NewTx(msg))
+}
diff --git a/op-service/txmgr/pair_test.go b/op-service/txmgr/pair_test.go
new file mode 100644
index 00000000000..a7b18de3c57
--- /dev/null
+++ b/op-service/txmgr/pair_test.go
@@ -0,0 +1,282 @@
+package txmgr
+
+import (
+	"context"
+	"slices"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/stretchr/testify/require"
+
+	"github.com/ethereum-optimism/optimism/op-service/testlog"
+	"github.com/ethereum-optimism/optimism/op-service/txmgr/metrics"
+)
+
+// pairTestHarness runs a real SimpleTxManager against the mock backend and
+// records every broadcast so tests can assert on cancellation no-ops and nonce
+// hygiene — the failure modes SendPairAsync exists to contain.
+type pairTestHarness struct {
+	mgr     *SimpleTxManager
+	backend *mockBackendWithNonce
+
+	mu         sync.Mutex
+	broadcasts []*types.Transaction
+}
+
+// pairLegBehavior tells the harness's tx sender what to do with a leg,
+// identified by its first data byte.
+type pairLegBehavior struct {
+	sendErr    error // returned from every SendTransaction call for this leg
+	mine       bool  // mine the tx (at its current fees) as soon as it is sent
+	mineStatus uint64
+}
+
+func newPairTestHarness(t *testing.T, behaviors map[byte]pairLegBehavior) *pairTestHarness {
+	backend := newMockBackendWithNonce(newGasPricer(1))
+	conf := configWithNumConfs(1)
+	conf.Backend = backend
+
+	mgr, err := NewSimpleTxManagerFromConfig("TEST", testlog.Logger(t, log.LevelCrit), &metrics.NoopTxMetrics{}, conf)
+	require.NoError(t, err)
+
+	h := &pairTestHarness{mgr: mgr, backend: backend}
+	backend.setTxSender(func(ctx context.Context, tx *types.Transaction) error {
+		h.mu.Lock()
+		h.broadcasts = append(h.broadcasts, tx)
+		h.mu.Unlock()
+
+		// Cancellation no-ops (empty-data self-sends crafted by
+		// cancelPairNonce) are always accepted and mined immediately.
+		if len(tx.Data()) == 0 {
+			hash := tx.Hash()
+			backend.mine(&hash, tx.GasFeeCap(), nil)
+			return nil
+		}
+		b, ok := behaviors[tx.Data()[0]]
+		require.True(t, ok, "unexpected tx broadcast with marker %d", tx.Data()[0])
+		if b.sendErr != nil {
+			return b.sendErr
+		}
+		if b.mine {
+			hash := tx.Hash()
+			backend.mineWithStatus(&hash, tx.GasFeeCap(), nil, b.mineStatus)
+		}
+		return nil
+	})
+	return h
+}
+
+// noopsAt returns the nonces of broadcast cancellation no-ops (empty-data
+// self-sends), sorted.
+func (h *pairTestHarness) noopsAt() []uint64 {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	var nonces []uint64
+	for _, tx := range h.broadcasts {
+		if len(tx.Data()) == 0 && tx.To() != nil && *tx.To() == h.mgr.cfg.From {
+			nonces = append(nonces, tx.Nonce())
+		}
+	}
+	slices.Sort(nonces)
+	return slices.Compact(nonces)
+}
+
+func pairCandidate(marker byte) TxCandidate {
+	return TxCandidate{TxData: []byte{marker}, To: &common.Address{0xbb}}
+}
+
+func sendPair(mgr *SimpleTxManager) (SendResponse, SendResponse) {
+	ch1 := make(chan SendResponse, 1)
+	ch2 := make(chan SendResponse, 1)
+	mgr.SendPairAsync(context.Background(), pairCandidate(1), pairCandidate(2), ch1, ch2)
+	return <-ch1, <-ch2
+}
+
+// mockBackendWithNonce initializes the manager's nonce from len(minedTxs), so
+// the first pair takes nonces 0 and 1 and a follow-up send takes 2.
+const pairBaseNonce = uint64(0)
+
+// TestSendPair_Success: both legs mine; consecutive nonces in argument order;
+// no cancellation no-ops are broadcast.
+func TestSendPair_Success(t *testing.T) {
+	h := newPairTestHarness(t, map[byte]pairLegBehavior{
+		1: {mine: true, mineStatus: types.ReceiptStatusSuccessful},
+		2: {mine: true, mineStatus: types.ReceiptStatusSuccessful},
+	})
+
+	r1, r2 := sendPair(h.mgr)
+
+	require.NoError(t, r1.Err)
+	require.NoError(t, r2.Err)
+	require.Equal(t, pairBaseNonce, r1.Nonce)
+	require.Equal(t, pairBaseNonce+1, r2.Nonce, "pair legs must take consecutive nonces in argument order")
+	require.Empty(t, h.noopsAt(), "a healthy pair must not broadcast cancellation no-ops")
+}
+
+// TestSendPair_FirstLegSendFailure is the nonce-gap hazard the pair exists to
+// contain: the auth-style first leg fails permanently (a critical nonce error
+// — txmgr treats unknown RPC errors as transient and retries them forever, so
+// permanent failures are SendState critical errors, timeouts, or
+// cancellations) after the second leg was already broadcast at the next
+// nonce. The pair must (a) cancel the orphaned second leg with a no-op at its
+// exact nonce, (b) fill the first leg's own nonce hole the same way, and (c)
+// leave the manager unwedged: the next send gets the next fresh nonce with no
+// reset and confirms normally.
+func TestSendPair_FirstLegSendFailure(t *testing.T) {
+	h := newPairTestHarness(t, map[byte]pairLegBehavior{
+		1: {sendErr: core.ErrNonceTooLow},
+		2: {},                                                      // accepted into the pool, never mined (gapped behind leg 1)
+		3: {mine: true, mineStatus: types.ReceiptStatusSuccessful}, // follow-up send
+	})
+
+	r1, r2 := sendPair(h.mgr)
+
+	require.Error(t, r1.Err)
+	require.ErrorIs(t, r2.Err, ErrPairLegCancelled)
+	require.Equal(t, []uint64{pairBaseNonce, pairBaseNonce + 1}, h.noopsAt(),
+		"both pair nonces must be consumed by cancellation no-ops")
+
+	// No wedge and no nonce reset: the next send continues the sequence at the
+	// nonce after the repaired pair and confirms.
+	ch := make(chan SendResponse, 1)
+	h.mgr.SendAsync(context.Background(), pairCandidate(3), ch)
+	r3 := <-ch
+	require.NoError(t, r3.Err)
+	require.Equal(t, pairBaseNonce+2, r3.Nonce)
+}
+
+// TestSendPair_FirstLegReverted: a mined-but-reverted first leg also fails the
+// pair (its event never fired), so the second leg is cancelled — but the first
+// leg's nonce was consumed on chain, so only the second leg's nonce gets a
+// no-op. The first leg's response keeps SendAsync semantics: a receipt with
+// failed status and no error.
+func TestSendPair_FirstLegReverted(t *testing.T) {
+	h := newPairTestHarness(t, map[byte]pairLegBehavior{
+		1: {mine: true, mineStatus: types.ReceiptStatusFailed},
+		2: {},                                                      // accepted, never mined; must be cancelled
+		3: {mine: true, mineStatus: types.ReceiptStatusSuccessful}, // follow-up send
+	})
+
+	r1, r2 := sendPair(h.mgr)
+
+	require.NoError(t, r1.Err, "a mined-but-reverted leg is a receipt, not a send error")
+	require.NotNil(t, r1.Receipt)
+	require.Equal(t, types.ReceiptStatusFailed, r1.Receipt.Status)
+	require.ErrorIs(t, r2.Err, ErrPairLegCancelled)
+	require.Equal(t, []uint64{pairBaseNonce + 1}, h.noopsAt(),
+		"only the unmined second leg's nonce needs a cancellation no-op")
+
+	ch := make(chan SendResponse, 1)
+	h.mgr.SendAsync(context.Background(), pairCandidate(3), ch)
+	r3 := <-ch
+	require.NoError(t, r3.Err)
+	require.Equal(t, pairBaseNonce+2, r3.Nonce)
+}
+
+// TestQueue_SendPair_Pipelines proves a pair holds exactly one maxPending
+// slot, in both directions. At most one: with maxPending=2, two pairs (four
+// txs) are all broadcast before anything confirms — per-tx slot accounting or
+// per-pair serialization would block this. At least one: a third pair is not
+// admitted (SendPair stays blocked, no nonces assigned) until a first pair
+// confirms and frees its slot.
+func TestQueue_SendPair_Pipelines(t *testing.T) {
+	const maxPending = 2 // queue slots
+	const numPairs = 2   // one slot each; 2*numPairs txs in flight at once
+	const numTxs = 2 * numPairs
+
+	wg := sync.WaitGroup{}
+	backend := newMockBackendWithConfirmationDelay(newGasPricer(3), &wg)
+	conf := configWithNumConfs(1)
+	conf.Backend = backend
+	// Keep resubmission fee-bumps out of the test's timing windows: a bumped
+	// tx is a new hash, which the delay backend would count against the
+	// WaitGroup unexpectedly.
+	conf.RebroadcastInterval.Store(int64(time.Minute))
+	conf.ResubmissionTimeout.Store(int64(time.Minute))
+	mgr, err := NewSimpleTxManagerFromConfig("TEST", testlog.Logger(t, log.LevelCrit), &metrics.NoopTxMetrics{}, conf)
+	require.NoError(t, err)
+
+	q := NewQueue[int](context.Background(), mgr, maxPending)
+
+	emptyCandidate := TxCandidate{To: &common.Address{}}
+
+	receiptChs := make([]chan TxReceipt[int], numTxs)
+	wg.Add(numTxs)
+	for pair := range numPairs {
+		id := 2 * pair
+		receiptChs[id] = make(chan TxReceipt[int], 1)
+		receiptChs[id+1] = make(chan TxReceipt[int], 1)
+		q.SendPair(
+			id, emptyCandidate, receiptChs[id],
+			id+1, emptyCandidate, receiptChs[id+1],
+		)
+	}
+
+	// cachedNonces returns the sorted nonces of every tx the backend has seen;
+	// nonceSeq builds the expected contiguous sequence from startingNonce.
+	cachedNonces := func() []uint64 {
+		nonces := make([]uint64, 0, len(backend.cachedTxs))
+		for _, tx := range backend.cachedTxs {
+			nonces = append(nonces, tx.Nonce())
+		}
+		slices.Sort(nonces)
+		return nonces
+	}
+	nonceSeq := func(n int) []uint64 {
+		seq := make([]uint64, n)
+		for i := range seq {
+			seq[i] = startingNonce + uint64(i)
+		}
+		return seq
+	}
+
+	// All four txs reach the backend while nothing has confirmed yet: the two
+	// pairs are genuinely in flight concurrently.
+	wg.Wait()
+	require.Equal(t, nonceSeq(numTxs), cachedNonces())
+
+	// Both slots are held, so a third pair must block in SendPair without
+	// being admitted (and without nonces assigned — they would land at +4/+5
+	// only after admission).
+	var thirdAdmitted atomic.Bool
+	thirdCh1 := make(chan TxReceipt[int], 1)
+	thirdCh2 := make(chan TxReceipt[int], 1)
+	wg.Add(2)
+	go func() {
+		q.SendPair(
+			numTxs, emptyCandidate, thirdCh1,
+			numTxs+1, emptyCandidate, thirdCh2,
+		)
+		thirdAdmitted.Store(true)
+	}()
+	require.Never(t, thirdAdmitted.Load, 1*time.Second, 10*time.Millisecond,
+		"a third pair was admitted while both slots were held: a pair must cost one slot")
+
+	// Confirming the first two pairs frees their slots; the third pair is then
+	// admitted and broadcasts at the next nonces.
+	backend.MineAll()
+	require.Eventually(t, thirdAdmitted.Load, 10*time.Second, 10*time.Millisecond,
+		"third pair was not admitted after slots freed")
+	wg.Wait() // third pair's txs reach the backend
+	backend.MineAll()
+	require.NoError(t, q.Wait())
+
+	for i, ch := range append(receiptChs, thirdCh1, thirdCh2) {
+		select {
+		case r := <-ch:
+			require.NoError(t, r.Err, "receipt %d", i)
+		case <-time.After(10 * time.Second):
+			t.Fatalf("timed out waiting for receipt %d", i)
+		}
+	}
+
+	// The third pair's nonces continue the sequence, assigned only after
+	// admission.
+	require.Equal(t, nonceSeq(numTxs+2), cachedNonces())
+}
diff --git a/op-service/txmgr/queue.go b/op-service/txmgr/queue.go
index 220b98567c1..da9823ebcde 100644
--- a/op-service/txmgr/queue.go
+++ b/op-service/txmgr/queue.go
@@ -2,6 +2,7 @@ package txmgr
 
 import (
 	"context"
+	"errors"
 	"math"
 	"sync"
 
@@ -83,6 +84,25 @@ func (q *Queue[T]) Send(id T, candidate TxCandidate, receiptCh chan TxReceipt[T]
 	q.txMgr.SendAsync(ctx, candidate, responseChan) // Nonce management handled synchronously, i.e. before this returns
 }
 
+// SendPair sends an ordered pair of transactions — see TxManager.SendPairAsync
+// for the pair contract — while holding a single maxPending slot for the whole
+// pair, so pairs pipeline at the same depth as individual txs sent via Send.
+//
+// Exactly one receipt is forwarded on each receipt channel. Like Send, this
+// blocks while the queue is at its maxPending limit, and a pair failure
+// cancels the queue's shared error group.
+func (q *Queue[T]) SendPair(firstID T, first TxCandidate, firstCh chan TxReceipt[T], secondID T, second TxCandidate, secondCh chan TxReceipt[T]) {
+	group, ctx := q.groupContext()
+	r1 := make(chan SendResponse, 1)
+	r2 := make(chan SendResponse, 1)
+	group.Go(func() error { // one slot for the whole pair
+		err1 := handleResponse(ctx, r1, firstCh, firstID)
+		err2 := handleResponse(ctx, r2, secondCh, secondID)
+		return errors.Join(err1, err2)
+	})
+	q.txMgr.SendPairAsync(ctx, first, second, r1, r2) // nonces for both legs assigned synchronously, in order
+}
+
 // TrySend sends the next tx, but only if the number of pending txs is below the
 // max pending.
 //
diff --git a/op-service/txmgr/txmgr.go b/op-service/txmgr/txmgr.go
index d0a99606dac..a9bfb6ddbdd 100644
--- a/op-service/txmgr/txmgr.go
+++ b/op-service/txmgr/txmgr.go
@@ -91,6 +91,30 @@ type TxManager interface {
 	// the order of nonce increments.
 	SendAsync(ctx context.Context, candidate TxCandidate, ch chan SendResponse)
 
+	// SendPairAsync submits two transactions as an ordered pair: both are crafted
+	// synchronously with nonces assigned in argument order (so the first leg
+	// precedes the second on chain), then broadcast back-to-back without waiting
+	// for the first to confirm, so pairs pipeline like individual SendAsync calls.
+	//
+	// Use a pair when the second transaction is only valid if the first executes
+	// successfully. If the first leg fails permanently (send failure, or mined
+	// but reverted), the second leg is cancelled with a fee-bumped no-op at its
+	// exact nonce; cancellation is best-effort, so callers must tolerate an
+	// orphaned second leg still landing on chain. Any leg that fails without
+	// mining gets the same no-op treatment, so a failed pair never resets the
+	// nonce, never leaves a nonce gap behind, and never disturbs concurrently
+	// pending transactions at higher nonces.
+	//
+	// Both response channels must be buffered; exactly one response is delivered
+	// on each, and only once both nonces are consumed on chain (repairs
+	// included). Repairs survive cancellation of ctx (bounded by TxSendTimeout
+	// when configured); if ctx is already cancelled no repair is attempted and
+	// in-flight legs are left to resolve in the pool.
+	//
+	// Blob candidates are not supported: the cancellation no-op cannot replace a
+	// blob transaction.
+	SendPairAsync(ctx context.Context, first TxCandidate, second TxCandidate, firstCh chan SendResponse, secondCh chan SendResponse)
+
 	// ChainID returns the chain this tx-manager is connected to
 	ChainID() eth.ChainID
 
diff --git a/op-service/txmgr/txmgr_test.go b/op-service/txmgr/txmgr_test.go
index 9b1020c876b..f070ba65be5 100644
--- a/op-service/txmgr/txmgr_test.go
+++ b/op-service/txmgr/txmgr_test.go
@@ -217,6 +217,7 @@ type minedTxInfo struct {
 	gasFeeCap   *big.Int
 	blobFeeCap  *big.Int
 	blockNumber uint64
+	status      uint64
 }
 
 // mockBackend implements ReceiptSource that tracks mined transactions
@@ -251,6 +252,12 @@ func (b *mockBackend) setTxSender(s sendTransactionFunc) {
 // TransactionReceipt with a matching txHash will result in a non-nil receipt.
 // If a nil txHash is supplied this has the effect of mining an empty block.
 func (b *mockBackend) mine(txHash *common.Hash, gasFeeCap, blobFeeCap *big.Int) {
+	b.mineWithStatus(txHash, gasFeeCap, blobFeeCap, types.ReceiptStatusSuccessful)
+}
+
+// mineWithStatus is mine with an explicit receipt execution status, to
+// simulate mined-but-reverted transactions.
+func (b *mockBackend) mineWithStatus(txHash *common.Hash, gasFeeCap, blobFeeCap *big.Int, status uint64) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
@@ -260,6 +267,7 @@ func (b *mockBackend) mine(txHash *common.Hash, gasFeeCap, blobFeeCap *big.Int)
 			gasFeeCap:   gasFeeCap,
 			blobFeeCap:  blobFeeCap,
 			blockNumber: b.blockHeight,
+			status:      status,
 		}
 	}
 }
@@ -349,6 +357,7 @@ func (b *mockBackend) TransactionReceipt(ctx context.Context, txHash common.Hash
 		GasUsed:           bigs.Uint64Strict(txInfo.gasFeeCap),
 		CumulativeGasUsed: blobFeeCap,
 		BlockNumber:       big.NewInt(int64(txInfo.blockNumber)),
+		Status:            txInfo.status,
 	}, nil
 }