diff --git a/docs/metrics.md b/docs/metrics.md
index 67c4ae8..bf1e8c1 100644
--- a/docs/metrics.md
+++ b/docs/metrics.md
@@ -168,8 +168,12 @@ Other useful starting conditions:
 
 `/healthz` checks liveness. Paid orchestrator `/readyz` requires writable payment
 storage, chain access, an active recipient, valid pricing, and worker completion.
-Checks run at heartbeat cadence even with empty queues. Dependency observations
-and worker completion expire after `max(60 seconds, 3 × HeartbeatInterval)`;
+Checks run at heartbeat cadence even with empty queues. Payment chain metadata
+refreshes every five seconds. An observation is refused at thirty seconds old,
+if the head timestamp is stale, or after thirty seconds without protocol L1
+progress. A successful RPC response with an unchanged head does not renew its
+age. One provider cannot prove absolute L1 lag. Other dependency checks and
+worker completion expire after `max(60 seconds, 3 × HeartbeatInterval)`;
 queue stalls and individual ticket errors remain separate signals.
 Off-chain orchestrator/discovery readiness reports listener health.
 Signer readiness requires chain access, positive deposit/total reserve, no
diff --git a/eth/payment.go b/eth/payment.go
index 196db67..8fb2f07 100644
--- a/eth/payment.go
+++ b/eth/payment.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"math/big"
+	"time"
 
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
@@ -15,7 +16,10 @@ import (
 
 // PaymentChain adapts the retained Livepeer contracts to pm's narrow chain
 // interface. Ethereum contract reads and redemption stay in eth.
-type PaymentChain struct{ Contracts *Contracts }
+type PaymentChain struct {
+	Contracts *Contracts
+	Observer  *PaymentObserver // Nil for direct transaction/redemption reads.
+}
 
 func (c PaymentChain) IsActive(ctx context.Context, recipient ethcommon.Address) (bool, error) {
 	snapshot, err := c.Snapshot(ctx)
@@ -26,6 +30,9 @@ func (c PaymentChain) IsActive(ctx context.Context, recipient ethcommon.Address)
 }
 
 func (c PaymentChain) IsActiveAt(ctx context.Context, recipient ethcommon.Address, snapshot ChainSnapshot) (bool, error) {
+	if err := snapshot.Validate(); err != nil {
+		return false, err
+	}
 	if snapshot.blockReference == nil {
 		return false, errors.New("canonical snapshot required")
 	}
@@ -37,7 +44,11 @@ func (c PaymentChain) IsActiveAt(ctx context.Context, recipient ethcommon.Addres
 	if err != nil {
 		return false, err
 	}
-	return bonding.IsActiveTranscoder(&bind.CallOpts{Context: ctx}, recipient)
+	active, err := bonding.IsActiveTranscoder(&bind.CallOpts{Context: ctx}, recipient)
+	if err == nil {
+		err = snapshot.Validate()
+	}
+	return active, err
 }
 
 // Receipt reports a submitted redemption without retrying or resubmitting it.
@@ -73,14 +84,29 @@ type ChainSnapshot struct {
 	Block          *big.Int // Protocol L1 block, including on Arbitrum.
 	Round          *big.Int // Last initialized round.
 	RoundHash      ethcommon.Hash
+	CurrentRound   *big.Int // Protocol round; may exceed the initialized Round.
 	blockReference map[string]any
+	observer       *PaymentObserver
+	generation     uint64
+	observedAt     time.Time
+	blockTime      time.Time
 }
 
 func (c PaymentChain) Snapshot(ctx context.Context) (ChainSnapshot, error) {
+	if c.Observer != nil {
+		return c.Observer.Snapshot(ctx)
+	}
+	at := time.Now()
 	header, err := c.Contracts.RPC.header(ctx, "latest")
 	if err != nil {
 		return ChainSnapshot{}, err
 	}
+	snapshot, err := c.snapshotAt(ctx, header, false)
+	snapshot.observedAt = at
+	return snapshot, err
+}
+
+func (c PaymentChain) snapshotAt(ctx context.Context, header *miniHeader, currentRound bool) (ChainSnapshot, error) {
 	if header.Hash == (ethcommon.Hash{}) || header.Number == nil {
 		return ChainSnapshot{}, errors.New("invalid chain header")
 	}
@@ -114,7 +140,17 @@ func (c PaymentChain) Snapshot(ctx context.Context) (ChainSnapshot, error) {
 	if hash == ([32]byte{}) {
 		return ChainSnapshot{}, errors.New("invalid round block hash")
 	}
-	return ChainSnapshot{Block: block, Round: round, RoundHash: ethcommon.Hash(hash), blockReference: ref}, nil
+	snapshot := ChainSnapshot{Block: block, Round: round, RoundHash: ethcommon.Hash(hash), blockReference: ref}
+	if currentRound {
+		snapshot.CurrentRound, err = rounds.CurrentRound(opts)
+		if err != nil {
+			return ChainSnapshot{}, err
+		}
+		if snapshot.CurrentRound.Cmp(round) < 0 {
+			return ChainSnapshot{}, errors.New("current round precedes initialized round")
+		}
+	}
+	return snapshot, nil
 }
 
 // SenderInfo is a coherent view of a sender's collateral and the amount of
@@ -153,7 +189,7 @@ func (c PaymentChain) SenderInfo(ctx context.Context, sender, recipient ethcommo
 			return SenderInfo{}, err
 		}
 	}
-	return SenderInfo{Snapshot: snapshot, Deposit: info.Sender.Deposit, Reserve: reserve, WithdrawRound: info.Sender.WithdrawRound}, nil
+	return SenderInfo{Snapshot: snapshot, Deposit: info.Sender.Deposit, Reserve: reserve, WithdrawRound: info.Sender.WithdrawRound}, snapshot.Validate()
 }
 
 // The redemption tuple follows Yondon Fu's go-livepeer/eth/client_ticketbroker.go
diff --git a/eth/payment_observer.go b/eth/payment_observer.go
new file mode 100644
index 0000000..44e5047
--- /dev/null
+++ b/eth/payment_observer.go
@@ -0,0 +1,203 @@
+// Adapted from livepeer/go-livepeer eth/watchers/timewatcher.go and
+// eth/blockwatch/block_watcher.go at 597f882cb9f4d42ad8430b0f592afd5929af4a6b:
+// Nico Vergauwen (b4605f705, 3aec3aa16), Yondon Fu (0acd88bbd), and
+// Rafał Leszko (4b6ede31f, 8c7ec0155). Immutable pinned observations and age
+// refusal also adapt Mike Zupper's BlueClawNetwork/livepeer-node eth/chainview.go
+// (2d95d40b8, 74f41b11f). Copyright (c) 2017 Livepeer; MIT notice in LICENSE.
+package eth
+
+import (
+	"context"
+	"errors"
+	"math/big"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+)
+
+var ErrObservationUnavailable = errors.New("chain observation unavailable")
+
+// PaymentObserver publishes complete canonical observations. Gaps and reorgs
+// rebuild state at the new hash instead of replaying uncertain event history.
+type PaymentObserver struct {
+	chain             PaymentChain
+	interval, maxAge  time.Duration
+	now               func() time.Time
+	mu                sync.RWMutex
+	current           ChainSnapshot
+	header            *miniHeader
+	generation        uint64
+	requireL1         bool
+	invalid, stopped  bool
+	contact, progress time.Time
+	l1Progress        time.Time
+	l1HighWater       *big.Int
+}
+
+func NewPaymentObserver(chain PaymentChain, requireL1 bool) *PaymentObserver {
+	chain.Observer = nil
+	return &PaymentObserver{chain: chain, interval: 5 * time.Second, maxAge: 30 * time.Second, now: time.Now, requireL1: requireL1 || chain.Contracts.RPC.requireL1}
+}
+
+// Run has one service-owned caller. Every RPC inherits the service lifetime.
+func (o *PaymentObserver) Run(ctx context.Context) {
+	defer func() { o.mu.Lock(); o.stopped = true; o.mu.Unlock() }()
+	ticker := time.NewTicker(o.interval)
+	defer ticker.Stop()
+	for {
+		read, cancel := context.WithTimeout(ctx, o.maxAge)
+		_ = o.refresh(read)
+		cancel()
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+	}
+}
+
+// ObserverStatus separates successful contact from accepted state and progress.
+// A single provider cannot establish an absolute bound on protocol L1 lag.
+type ObserverStatus struct {
+	ObservedAt, ContactAt, HeadProgressAt, L1ProgressAt time.Time
+	Healthy                                             bool
+}
+
+func (o *PaymentObserver) Status() ObserverStatus {
+	o.mu.RLock()
+	defer o.mu.RUnlock()
+	return ObserverStatus{o.current.observedAt, o.contact, o.progress, o.l1Progress, o.validate(o.current) == nil}
+}
+
+func (o *PaymentObserver) Snapshot(ctx context.Context) (ChainSnapshot, error) {
+	if err := ctx.Err(); err != nil {
+		return ChainSnapshot{}, err
+	}
+	o.mu.RLock()
+	defer o.mu.RUnlock()
+	if err := o.validate(o.current); err != nil {
+		return ChainSnapshot{}, err
+	}
+	s := o.current
+	s.Block, s.Round, s.CurrentRound = new(big.Int).Set(s.Block), new(big.Int).Set(s.Round), new(big.Int).Set(s.CurrentRound)
+	return s, nil
+}
+
+func (s ChainSnapshot) ObservedAt() time.Time {
+	if s.observedAt.IsZero() { // Injected direct chains have no retained observation.
+		return time.Now()
+	}
+	return s.observedAt
+}
+
+// Validate also rejects snapshots retained across an invalidation or shutdown.
+// Direct reads and injected test chains retain their existing behavior.
+func (s ChainSnapshot) Validate() error {
+	if s.observer == nil {
+		return nil
+	}
+	s.observer.mu.RLock()
+	defer s.observer.mu.RUnlock()
+	return s.observer.validate(s)
+}
+
+func (o *PaymentObserver) validate(s ChainSnapshot) error {
+	if o.stopped || o.invalid || s.generation != o.generation || o.expired(s, o.l1Progress) {
+		return ErrObservationUnavailable
+	}
+	return nil
+}
+
+func (o *PaymentObserver) expired(s ChainSnapshot, l1Progress time.Time) bool {
+	now := o.now()
+	return s.observedAt.IsZero() || now.Sub(s.observedAt) >= o.maxAge || now.Sub(s.blockTime) >= o.maxAge || now.Sub(l1Progress) >= o.maxAge
+}
+
+// Only Run calls refresh. Previous/current identity is the extension point for
+// complete hash-pinned logs or reset; no partial delta authorizes publication.
+func (o *PaymentObserver) refresh(ctx context.Context) error {
+	at := o.now()
+	head, err := o.chain.Contracts.RPC.header(ctx, "latest")
+	if err != nil {
+		return err
+	}
+	o.mu.Lock()
+	o.contact = o.now()
+	prior := o.header
+	gap := false
+	if head.Hash == (common.Hash{}) || head.Number == nil {
+		o.mu.Unlock()
+		return ErrObservationUnavailable
+	}
+	if prior != nil {
+		distance := new(big.Int).Sub((*big.Int)(head.Number), (*big.Int)(prior.Number))
+		if distance.Sign() < 0 {
+			o.invalid = true
+			o.generation++
+			o.mu.Unlock()
+			return ErrObservationUnavailable
+		}
+		gap = head.Hash != prior.Hash && distance.Cmp(big.NewInt(1)) > 0
+		if head.Hash != prior.Hash && !gap && (distance.Sign() == 0 || head.ParentHash != prior.Hash) {
+			o.invalid = true
+			o.generation++
+		}
+	}
+	// Usable identity conflicts invalidate even when other header metadata is
+	// missing or stale. Such metadata must not restore a rejected observation.
+	if head.Timestamp == nil || (o.requireL1 && head.L1BlockNumber == nil) {
+		o.mu.Unlock()
+		return ErrObservationUnavailable
+	}
+	blockTime := time.Unix(int64(*head.Timestamp), 0)
+	if at.Sub(blockTime) >= o.maxAge || blockTime.After(at.Add(o.maxAge)) {
+		o.mu.Unlock()
+		return ErrObservationUnavailable
+	}
+	if prior != nil && head.Hash == prior.Hash {
+		// A lagging backend may return N, N-1, then N again. Restore the
+		// immutable observation without renewing any freshness deadline.
+		o.current.generation, o.invalid = o.generation, false
+		err := o.validate(o.current)
+		o.mu.Unlock()
+		return err
+	}
+	o.mu.Unlock()
+	if gap {
+		// A polling gap does not itself orphan the retained observation. Prove
+		// its anchor before keeping it available during the pinned backfill.
+		anchor, err := o.chain.Contracts.RPC.header(ctx, hexutil.EncodeBig((*big.Int)(prior.Number)))
+		if err != nil || anchor.Hash != prior.Hash || anchor.Number == nil || (*big.Int)(anchor.Number).Cmp((*big.Int)(prior.Number)) != 0 {
+			o.mu.Lock()
+			o.invalid = true
+			o.generation++
+			o.mu.Unlock()
+		}
+	}
+	s, err := o.chain.snapshotAt(ctx, head, true)
+	if err != nil {
+		return err
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	progress, l1Progress, highWater := o.progress, o.l1Progress, o.l1HighWater
+	if prior == nil || (*big.Int)(head.Number).Cmp((*big.Int)(prior.Number)) > 0 {
+		progress = at
+	}
+	if highWater == nil || s.Block.Cmp(highWater) > 0 {
+		l1Progress, highWater = at, s.Block
+	}
+	s.observer, s.generation, s.observedAt, s.blockTime = o, o.generation, at, blockTime
+	if o.stopped || o.expired(s, l1Progress) {
+		return ErrObservationUnavailable
+	}
+	o.current, o.header, o.invalid = s, head, false
+	o.progress, o.l1Progress, o.l1HighWater = progress, l1Progress, highWater
+	o.requireL1 = o.requireL1 || head.L1BlockNumber != nil
+	return nil
+}
diff --git a/eth/payment_observer_test.go b/eth/payment_observer_test.go
new file mode 100644
index 0000000..1a1cd69
--- /dev/null
+++ b/eth/payment_observer_test.go
@@ -0,0 +1,339 @@
+package eth
+
+import (
+	"context"
+	"math/big"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/livepeer/node/internal/test"
+	"github.com/stretchr/testify/require"
+)
+
+// Initialization, round rollback and cancellation cases adapt go-livepeer
+// eth/watchers/timewatcher_test.go and eth/blockwatch/block_watcher_test.go at
+// 597f882cb9f4d42ad8430b0f592afd5929af4a6b, by Nico Vergauwen, Yondon Fu and
+// Rafał Leszko. Pinned-copy/expiry cases also adapt Mike Zupper's PR #1
+// eth/chainview_test.go at bbe126e. MIT notice in LICENSE.
+func newTestObserver(t *testing.T) (*test.ChainRPC, *PaymentObserver, *time.Time) {
+	t.Helper()
+	f := test.NewChainRPC(t)
+	rpc, err := OpenRPC(f.URL)
+	require.NoError(t, err)
+	t.Cleanup(rpc.Close)
+	c, err := NewContracts(rpc, f.Controller)
+	require.NoError(t, err)
+	o := NewPaymentObserver(PaymentChain{Contracts: c}, true)
+	now := time.Now().Truncate(time.Second)
+	o.now = func() time.Time { return now }
+	return f, o, &now
+}
+
+func advanceObserver(f *test.ChainRPC, now time.Time) {
+	f.Update(func(s *test.ChainState) {
+		s.Parent, s.Number, s.Timestamp = s.Hash, s.Number+1, uint64(now.Unix())
+		s.Hash = common.BigToHash(new(big.Int).SetUint64(s.Number))
+		s.L1Block++
+	})
+}
+
+func TestObserverLocalImmutableSnapshot(t *testing.T) {
+	f, o, _ := newTestObserver(t)
+	_, err := o.Snapshot(t.Context())
+	require.ErrorIs(t, err, ErrObservationUnavailable)
+	f.Update(func(s *test.ChainState) { s.CurrentRound = 6 })
+	require.NoError(t, o.refresh(t.Context()))
+	before := f.Stats()
+	for range 10 {
+		s, err := o.Snapshot(t.Context())
+		require.NoError(t, err)
+		require.Equal(t, int64(50), s.Block.Int64(), "L2 height is not the protocol clock")
+		require.Equal(t, int64(5), s.Round.Int64())
+		require.Equal(t, int64(6), s.CurrentRound.Int64())
+		s.Block.SetInt64(0)
+		s.Round.SetInt64(0)
+		s.CurrentRound.SetInt64(0)
+	}
+	require.Equal(t, before, f.Stats(), "warm snapshots and status must be local")
+	require.True(t, o.Status().Healthy)
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	_, err = o.Snapshot(ctx)
+	require.ErrorIs(t, err, context.Canceled)
+}
+
+func TestObserverExplicitArbitrumChainIDRequiresL1(t *testing.T) {
+	f, original, _ := newTestObserver(t)
+	f.Update(func(s *test.ChainState) { s.ChainID, s.OmitL1 = 42161, true })
+	rpc := original.chain.Contracts.RPC
+	require.NoError(t, rpc.CheckChainID(t.Context(), big.NewInt(42161)))
+	o := NewPaymentObserver(original.chain, false) // No named network preset.
+	require.ErrorIs(t, o.refresh(t.Context()), ErrObservationUnavailable)
+	// A non-Arbitrum custom chain continues to use its native block clock.
+	f.Update(func(s *test.ChainState) { s.ChainID = 1 })
+	rpc.requireL1 = false
+	require.NoError(t, rpc.CheckChainID(t.Context(), big.NewInt(1)))
+	o = NewPaymentObserver(original.chain, false)
+	require.NoError(t, o.refresh(t.Context()))
+}
+
+func TestObserverContactDoesNotRenewOldHead(t *testing.T) {
+	f, o, now := newTestObserver(t)
+	require.NoError(t, o.refresh(t.Context()))
+	initial := o.Status()
+	old, err := o.Snapshot(t.Context())
+	require.NoError(t, err)
+	*now = now.Add(10 * time.Second)
+	require.NoError(t, o.refresh(t.Context()))
+	status := o.Status()
+	require.Equal(t, initial.ObservedAt, status.ObservedAt)
+	require.Equal(t, initial.HeadProgressAt, status.HeadProgressAt)
+	require.True(t, status.ContactAt.After(initial.ContactAt))
+	original := f.State()
+	f.Update(func(s *test.ChainState) { s.Number-- })
+	require.Error(t, o.refresh(t.Context()))
+	f.Update(func(s *test.ChainState) { *s = original })
+	require.NoError(t, o.refresh(t.Context()))
+	require.True(t, o.Status().Healthy)
+	require.ErrorIs(t, old.Validate(), ErrObservationUnavailable)
+	require.Equal(t, initial.ObservedAt, o.Status().ObservedAt)
+	*now = now.Add(20 * time.Second)
+	require.Error(t, o.refresh(t.Context()))
+	_, err = o.Snapshot(t.Context())
+	require.ErrorIs(t, err, ErrObservationUnavailable)
+	advanceObserver(f, *now)
+	require.NoError(t, o.refresh(t.Context()))
+	require.True(t, o.Status().Healthy)
+	*now = now.Add(time.Second)
+	advanceObserver(f, now.Add(-time.Minute)) // A responsive, advancing lagging head.
+	require.Error(t, o.refresh(t.Context()))
+	require.Equal(t, status.ObservedAt.Add(30*time.Second), o.Status().ObservedAt)
+}
+
+func TestObserverInvalidHeaders(t *testing.T) {
+	for _, change := range []func(*test.ChainState){
+		func(s *test.ChainState) { s.OmitL1 = true },
+		func(s *test.ChainState) { s.RoundHash = common.Hash{} },
+		func(s *test.ChainState) { s.Round = 0 },
+		func(s *test.ChainState) { s.CurrentRound = 4 },
+		func(s *test.ChainState) { s.Timestamp += 60 },
+	} {
+		f, o, _ := newTestObserver(t)
+		f.Update(change)
+		require.Error(t, o.refresh(t.Context()))
+		require.False(t, o.Status().Healthy)
+	}
+}
+
+func TestObserverRejectedHeadersInvalidateConflictingIdentity(t *testing.T) {
+	for _, metadata := range []string{"stale", "missing timestamp", "missing L1"} {
+		for _, mode := range []string{"replacement", "rollback", "parent", "gap", "same"} {
+			t.Run(metadata+"/"+mode, func(t *testing.T) {
+				f, o, now := newTestObserver(t)
+				require.NoError(t, o.refresh(t.Context()))
+				held, err := o.Snapshot(t.Context())
+				require.NoError(t, err)
+				f.Update(func(s *test.ChainState) {
+					if mode != "same" {
+						s.Hash, s.Parent = common.HexToHash("0xdead"), common.HexToHash("0xbad")
+					}
+					switch mode {
+					case "rollback":
+						s.Number--
+					case "parent":
+						s.Number++
+					case "gap":
+						s.Number += 10
+					}
+					switch metadata {
+					case "stale":
+						s.Timestamp = uint64(now.Add(-31 * time.Second).Unix())
+					case "missing timestamp":
+						s.OmitTimestamp = true
+					case "missing L1":
+						s.OmitL1 = true
+					}
+				})
+				f.Reset()
+				require.ErrorIs(t, o.refresh(t.Context()), ErrObservationUnavailable)
+				if mode == "gap" || mode == "same" {
+					require.NoError(t, held.Validate(), "rejected metadata alone cannot orphan retained state")
+					require.True(t, o.Status().Healthy)
+				} else {
+					require.ErrorIs(t, held.Validate(), ErrObservationUnavailable)
+					require.False(t, o.Status().Healthy)
+					_, err = o.Snapshot(t.Context())
+					require.ErrorIs(t, err, ErrObservationUnavailable)
+				}
+				require.Equal(t, held.ObservedAt(), o.Status().ObservedAt)
+				require.Equal(t, 1, f.Stats().RoundTrips, "conflicts must invalidate before any further RPC")
+			})
+		}
+	}
+}
+
+func TestObserverL1ProgressDeadline(t *testing.T) {
+	for _, clocks := range [][]uint64{{50, 50, 50}, {49, 50, 49}} {
+		f, o, now := newTestObserver(t)
+		require.NoError(t, o.refresh(t.Context()))
+		progress, accepted := o.Status().L1ProgressAt, o.Status().ObservedAt
+		for step, clock := range clocks {
+			*now = now.Add(10 * time.Second)
+			advanceObserver(f, *now)
+			f.Update(func(s *test.ChainState) { s.L1Block = clock })
+			err := o.refresh(t.Context())
+			if step < 2 {
+				require.NoError(t, err)
+				accepted = o.Status().ObservedAt
+			} else {
+				require.ErrorIs(t, err, ErrObservationUnavailable)
+				require.Equal(t, accepted, o.Status().ObservedAt)
+			}
+			require.Equal(t, progress, o.Status().L1ProgressAt)
+		}
+		require.False(t, o.Status().Healthy)
+		advanceObserver(f, *now)
+		f.Update(func(s *test.ChainState) { s.L1Block = 51 })
+		require.NoError(t, o.refresh(t.Context()))
+		require.True(t, o.Status().Healthy)
+	}
+}
+
+func TestObserverExpiredCandidateIsNotPublished(t *testing.T) {
+	f, o, now := newTestObserver(t)
+	require.NoError(t, o.refresh(t.Context()))
+	old := o.Status()
+	advanceObserver(f, *now)
+	f.Update(func(s *test.ChainState) {
+		s.Before = func(_ context.Context, name string) {
+			if name == "RoundsManager.currentRound" {
+				*now = now.Add(30 * time.Second)
+			}
+		}
+	})
+	require.ErrorIs(t, o.refresh(t.Context()), ErrObservationUnavailable)
+	status := o.Status()
+	require.Equal(t, old.ObservedAt, status.ObservedAt)
+	require.Equal(t, old.HeadProgressAt, status.HeadProgressAt)
+	require.Equal(t, old.L1ProgressAt, status.L1ProgressAt)
+	require.Equal(t, int64(50), o.current.Block.Int64())
+}
+
+func TestObserverPinnedBackfillAndReorg(t *testing.T) {
+	f, o, now := newTestObserver(t)
+	require.NoError(t, o.refresh(t.Context()))
+	old, err := o.Snapshot(t.Context())
+	require.NoError(t, err)
+	advanceObserver(f, *now)
+	f.Update(func(s *test.ChainState) { s.Fail = "RoundsManager.blockHashForRound"; s.Round, s.CurrentRound = 6, 7 })
+	require.Error(t, o.refresh(t.Context()))
+	s, err := o.Snapshot(t.Context())
+	require.NoError(t, err)
+	require.Equal(t, old.Round, s.Round, "failed candidates publish no partial round state")
+	f.Update(func(s *test.ChainState) { s.Number += 10; s.Hash = common.HexToHash("0xbeef"); s.Fail = "" })
+	var once sync.Once
+	f.Update(func(s *test.ChainState) {
+		s.Before = func(_ context.Context, name string) {
+			if name == "RoundsManager.blockHashForRound" {
+				once.Do(func() { f.Update(func(s *test.ChainState) { s.Hash = common.HexToHash("0xbad") }) })
+			}
+		}
+	})
+	require.Error(t, o.refresh(t.Context()), "a reorg during pinned reads rejects the candidate")
+	require.NoError(t, s.Validate(), "an orphaned candidate does not orphan its proven canonical ancestor")
+	f.Update(func(s *test.ChainState) { s.Before = nil })
+	require.NoError(t, o.refresh(t.Context()))
+}
+
+func TestObserverCancellationJoinsRPC(t *testing.T) {
+	f, o, _ := newTestObserver(t)
+	o.now = time.Now
+	entered, exited := make(chan struct{}), make(chan struct{})
+	f.Update(func(s *test.ChainState) {
+		s.Before = func(ctx context.Context, _ string) {
+			close(entered)
+			<-ctx.Done()
+			close(exited)
+		}
+	})
+	ctx, cancel := context.WithCancel(t.Context())
+	done := make(chan struct{})
+	go func() { o.Run(ctx); close(done) }()
+	select {
+	case <-entered:
+	case <-time.After(time.Second):
+		t.Fatal("observer did not start")
+	}
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("observer did not stop")
+	}
+	select {
+	case <-exited:
+	case <-time.After(time.Second):
+		t.Fatal("RPC outlived observer cancellation")
+	}
+	require.False(t, o.Status().Healthy)
+	require.Equal(t, 1, f.Stats().RoundTrips)
+}
+
+func TestObserverGapKeepsCanonicalStateAvailable(t *testing.T) {
+	f, o, _ := newTestObserver(t)
+	require.NoError(t, o.refresh(t.Context()))
+	for _, mode := range []string{"gap", "replacement", "parent"} {
+		old, err := o.Snapshot(t.Context())
+		require.NoError(t, err)
+		entered, unblock := make(chan struct{}), make(chan struct{})
+		notify := sync.OnceFunc(func() { close(entered) })
+		release := sync.OnceFunc(func() { close(unblock) })
+		t.Cleanup(release)
+		f.Update(func(s *test.ChainState) {
+			if mode == "gap" {
+				s.Number += 10
+				s.L1Block++
+			}
+			if mode == "parent" {
+				s.Number++
+			}
+			s.Hash = common.BigToHash(big.NewInt(int64(s.Number) + int64(s.L1Block)))
+			if mode == "replacement" {
+				s.Hash = common.HexToHash("0xf00d")
+				s.Round, s.CurrentRound = 4, 5
+			}
+			s.Before = func(_ context.Context, name string) {
+				if name == "RoundsManager.lastInitializedRound" {
+					notify()
+					<-unblock
+				}
+			}
+		})
+		f.Reset()
+		result := make(chan error, 1)
+		go func() { result <- o.refresh(t.Context()) }()
+		select {
+		case <-entered:
+		case <-time.After(time.Second):
+			t.Fatal("backfill did not start")
+		}
+		if mode != "gap" {
+			require.ErrorIs(t, old.Validate(), ErrObservationUnavailable)
+			require.False(t, o.Status().Healthy)
+			require.Equal(t, 1, f.Stats().Calls["eth_getBlockByNumber"], "proven reorgs must invalidate before any anchor probe")
+		} else {
+			require.NoError(t, old.Validate())
+			require.True(t, o.Status().Healthy)
+			_, err = o.Snapshot(t.Context())
+			require.NoError(t, err)
+		}
+		release()
+		require.NoError(t, <-result)
+		f.Update(func(s *test.ChainState) { s.Before = nil })
+		if mode == "replacement" {
+			require.Equal(t, int64(4), o.current.Round.Int64(), "removed round state is backfilled")
+		}
+	}
+}
diff --git a/eth/payment_test.go b/eth/payment_test.go
index 6a9018d..4ccaaed 100644
--- a/eth/payment_test.go
+++ b/eth/payment_test.go
@@ -96,7 +96,10 @@ func TestCanonicalPaymentSnapshotAndCollateral(t *testing.T) {
 			require.True(t, active)
 			funds, err := chain.SenderInfo(t.Context(), sender, recipient)
 			require.NoError(t, err)
-			require.Equal(t, snapshot, funds.Snapshot)
+			require.Equal(t, snapshot.Block, funds.Snapshot.Block)
+			require.Equal(t, snapshot.Round, funds.Snapshot.Round)
+			require.Equal(t, snapshot.RoundHash, funds.Snapshot.RoundHash)
+			require.Equal(t, snapshot.blockReference, funds.Snapshot.blockReference)
 			require.Equal(t, big.NewInt(10), funds.Deposit)
 			require.Equal(t, big.NewInt(9), funds.WithdrawRound)
 			require.Equal(t, big.NewInt(7), funds.Reserve, "acceptance uses this recipient's claimable reserve")
diff --git a/eth/rpc.go b/eth/rpc.go
index 36ad348..d023838 100644
--- a/eth/rpc.go
+++ b/eth/rpc.go
@@ -26,9 +26,10 @@ const maxRPCResponse = 1 << 20
 
 // RPC applies bounded HTTP transport and error redaction to geth's client.
 type RPC struct {
-	ethereum *ethclient.Client
-	http     *http.Client
-	observer func(string, time.Duration, error)
+	ethereum  *ethclient.Client
+	http      *http.Client
+	observer  func(string, time.Duration, error)
+	requireL1 bool // Set by the startup chain-ID check, before concurrent use.
 }
 
 // SetObserver installs a hook before the client is used. Observations contain
@@ -219,10 +220,12 @@ func (r *RPC) TransactionReceipt(ctx context.Context, hash ethcommon.Hash) (*typ
 // miniHeader retains the RPC hash and Arbitrum's L1 clock, which geth's Header
 // does not expose. All quantities use geth's JSON decoders.
 type miniHeader struct {
-	Number        *hexutil.Big   `json:"number"`
-	BaseFee       *hexutil.Big   `json:"baseFeePerGas"`
-	L1BlockNumber *hexutil.Big   `json:"l1BlockNumber"`
-	Hash          ethcommon.Hash `json:"hash"`
+	Number        *hexutil.Big    `json:"number"`
+	BaseFee       *hexutil.Big    `json:"baseFeePerGas"`
+	L1BlockNumber *hexutil.Big    `json:"l1BlockNumber"`
+	Hash          ethcommon.Hash  `json:"hash"`
+	ParentHash    ethcommon.Hash  `json:"parentHash"`
+	Timestamp     *hexutil.Uint64 `json:"timestamp"`
 }
 
 func (r *RPC) header(ctx context.Context, block any) (*miniHeader, error) {
@@ -245,5 +248,11 @@ func (r *RPC) CheckChainID(ctx context.Context, expected *big.Int) error {
 	if expected != nil && id.Cmp(expected) != 0 {
 		return errors.New("RPC chain ID does not match configured chain-id")
 	}
+	if id.IsUint64() {
+		switch id.Uint64() {
+		case 42161, 42170, 421613, 421614: // Arbitrum One, Nova, Goerli, Sepolia.
+			r.requireL1 = true
+		}
+	}
 	return nil
 }
diff --git a/internal/test/chainrpc.go b/internal/test/chainrpc.go
index 1b8cdd7..1738f27 100644
--- a/internal/test/chainrpc.go
+++ b/internal/test/chainrpc.go
@@ -1,14 +1,17 @@
 package test
 
 import (
+	"context"
 	"encoding/hex"
 	"encoding/json"
+	"fmt"
 	"maps"
 	"net/http"
 	"net/http/httptest"
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
@@ -17,13 +20,25 @@ import (
 
 // Adapted from BlueClawNetwork/livepeer-node internal/test/chainrpc.go,
 // 285eda984d7cce4e3030ef0510d63ae142534dd7, by Mike Zupper (MIT; see LICENSE).
-// ChainRPC serves the payment read methods at one canonical block. Other tests
+// ChainRPC serves payment reads at canonical blocks. Other tests
 // can extend this fixture without duplicating the payment RPC protocol.
 type ChainRPC struct {
 	URL        string
 	Controller common.Address
 	mu         sync.Mutex
 	stats      ChainRPCStats
+	state      ChainState
+	history    map[common.Hash]ChainState
+}
+
+type ChainState struct {
+	Number, L1Block, Timestamp uint64
+	ChainID                    uint64
+	Hash, Parent, RoundHash    common.Hash
+	Round, CurrentRound        uint64
+	OmitL1, OmitTimestamp      bool
+	Fail                       string
+	Before                     func(context.Context, string)
 }
 
 type ChainRPCStats struct {
@@ -34,7 +49,8 @@ type ChainRPCStats struct {
 func NewChainRPC(t testing.TB) *ChainRPC {
 	t.Helper()
 	c := &ChainRPC{Controller: common.HexToAddress("0xc001"), stats: ChainRPCStats{Calls: map[string]int{}}}
-	blockHash := common.HexToHash("0xabcd").Hex()
+	c.state = ChainState{Number: 0x100000, L1Block: 50, Timestamp: uint64(time.Now().Unix()), Hash: common.HexToHash("0xabcd"), RoundHash: common.HexToHash("0x1234"), Round: 5, CurrentRound: 5}
+	c.history = map[common.Hash]ChainState{c.state.Hash: c.state}
 	word := func(value string) string { return strings.Repeat("0", 64-len(value)) + value }
 	addresses := map[string]string{}
 	for name, address := range map[string]string{"RoundsManager": "a1", "BondingManager": "a2", "TicketBroker": "a3"} {
@@ -43,6 +59,7 @@ func NewChainRPC(t testing.TB) *ChainRPC {
 	methods := map[string]string{}
 	for signature, name := range map[string]string{
 		"getContract(bytes32)": "Controller.getContract", "lastInitializedRound()": "RoundsManager.lastInitializedRound",
+		"currentRound()":             "RoundsManager.currentRound",
 		"blockHashForRound(uint256)": "RoundsManager.blockHashForRound", "isActiveTranscoder(address)": "BondingManager.isActiveTranscoder",
 		"getSenderInfo(address)": "TicketBroker.getSenderInfo", "claimableReserve(address,address)": "TicketBroker.claimableReserve",
 	} {
@@ -56,27 +73,71 @@ func NewChainRPC(t testing.TB) *ChainRPC {
 		}
 		require.NoError(t, json.NewDecoder(r.Body).Decode(&request))
 		name := request.Method
+		var call struct{ Input string }
+		if name == "eth_call" {
+			require.Len(t, request.Params, 2)
+			require.NoError(t, json.Unmarshal(request.Params[0], &call))
+			require.GreaterOrEqual(t, len(call.Input), 10)
+			name = methods[call.Input[2:10]]
+		}
+		c.mu.Lock()
+		c.stats.RoundTrips++
+		c.stats.Calls[name]++
+		before := c.state.Before
+		c.mu.Unlock()
+		if before != nil {
+			before(r.Context(), name)
+		}
+		state := c.State()
+		fail := func() {
+			_ = json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": request.ID, "error": map[string]any{"code": -32000}})
+		}
+		if state.Fail == name {
+			fail()
+			return
+		}
 		var result any
 		switch request.Method {
+		case "eth_chainId":
+			result = quantity(state.ChainID)
 		case "eth_getBlockByNumber":
-			result = map[string]string{"number": "0x100000", "l1BlockNumber": "0x32", "hash": blockHash}
+			var ref string
+			require.NoError(t, json.Unmarshal(request.Params[0], &ref))
+			if ref != "latest" {
+				var ok bool
+				state, ok = c.stateAt(ref)
+				if !ok {
+					break
+				}
+			}
+			header := map[string]any{"number": quantity(state.Number), "hash": state.Hash, "parentHash": state.Parent}
+			if !state.OmitTimestamp {
+				header["timestamp"] = quantity(state.Timestamp)
+			}
+			if !state.OmitL1 {
+				header["l1BlockNumber"] = quantity(state.L1Block)
+			}
+			result = header
 		case "eth_call":
-			require.Len(t, request.Params, 2)
 			var ref map[string]any
 			require.NoError(t, json.Unmarshal(request.Params[1], &ref))
-			require.Equal(t, map[string]any{"blockHash": blockHash, "requireCanonical": true}, ref)
-			var call struct{ Input string }
-			require.NoError(t, json.Unmarshal(request.Params[0], &call))
-			require.GreaterOrEqual(t, len(call.Input), 10)
-			name = methods[call.Input[2:10]]
+			require.Equal(t, true, ref["requireCanonical"])
+			var ok bool
+			state, ok = c.stateAt(ref["blockHash"].(string))
+			if !ok {
+				fail()
+				return
+			}
 			switch name {
 			case "Controller.getContract":
 				require.Contains(t, addresses, call.Input[10:])
 				result = "0x" + addresses[call.Input[10:]]
 			case "RoundsManager.lastInitializedRound":
-				result = "0x" + word("5")
+				result = "0x" + word(quantity(state.Round)[2:])
+			case "RoundsManager.currentRound":
+				result = "0x" + word(quantity(state.CurrentRound)[2:])
 			case "RoundsManager.blockHashForRound":
-				result = common.HexToHash("0x1234").Hex()
+				result = state.RoundHash.Hex()
 			case "BondingManager.isActiveTranscoder":
 				result = "0x" + word("1")
 			case "TicketBroker.getSenderInfo":
@@ -89,10 +150,6 @@ func NewChainRPC(t testing.TB) *ChainRPC {
 		default:
 			t.Errorf("unexpected payment RPC method %q", request.Method)
 		}
-		c.mu.Lock()
-		c.stats.RoundTrips++
-		c.stats.Calls[name]++
-		c.mu.Unlock()
 		require.NoError(t, json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": result}))
 	}))
 	t.Cleanup(server.Close)
@@ -100,8 +157,45 @@ func NewChainRPC(t testing.TB) *ChainRPC {
 	return c
 }
 
+func (c *ChainRPC) State() ChainState {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.state
+}
+
+func (c *ChainRPC) Update(update func(*ChainState)) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	update(&c.state)
+	for h, state := range c.history {
+		if state.Number >= c.state.Number {
+			delete(c.history, h)
+		}
+	}
+	c.history[c.state.Hash] = c.state
+}
+
+func (c *ChainRPC) stateAt(ref string) (ChainState, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for _, state := range c.history {
+		if ref == state.Hash.Hex() || ref == quantity(state.Number) {
+			return state, true
+		}
+	}
+	return ChainState{}, false
+}
+
+func quantity(n uint64) string { return fmt.Sprintf("0x%x", n) }
+
 func (c *ChainRPC) Stats() ChainRPCStats {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	return ChainRPCStats{RoundTrips: c.stats.RoundTrips, Calls: maps.Clone(c.stats.Calls)}
 }
+
+func (c *ChainRPC) Reset() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.stats = ChainRPCStats{Calls: map[string]int{}}
+}
diff --git a/orchestrator/app.go b/orchestrator/app.go
index c066b7c..d182656 100644
--- a/orchestrator/app.go
+++ b/orchestrator/app.go
@@ -260,6 +260,7 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) {
 	var engine *PaymentEngine
 	var redeemerDB *RedeemerDB
 	var paymentChain eth.PaymentChain
+	var observedChain eth.PaymentChain
 	var redeemerKey *eth.Key
 	var paymentChainID *big.Int
 	if p.paymentsRequested() {
@@ -291,6 +292,8 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) {
 			contracts.MaxFeePerGas = p.RedeemerMaxFeePerGas.ToBig()
 		}
 		paymentChain = eth.PaymentChain{Contracts: contracts}
+		observedChain = paymentChain
+		observedChain.Observer = eth.NewPaymentObserver(paymentChain, p.Network == nodeconfig.Mainnet)
 		if p.ETHUSDFeed != nil {
 			rate, until, err := contracts.WeiPerUSD(parent, *p.ETHUSDFeed, p.PriceMaxAge)
 			if err != nil {
@@ -304,7 +307,7 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) {
 		if p.OrchestratorAddress != nil {
 			recipient = *p.OrchestratorAddress
 		}
-		engine, err = NewPaymentEngine(redeemerDB, pm.EthereumChain{Client: paymentChain}, recipient, p.TicketFaceValue.ToBig(), p.TicketWinProb.ToBig())
+		engine, err = NewPaymentEngine(redeemerDB, pm.EthereumChain{Client: observedChain}, recipient, p.TicketFaceValue.ToBig(), p.TicketWinProb.ToBig())
 		if err != nil {
 			return fmt.Errorf("initialize payment engine (check --account, --ticket-face-value and --ticket-win-prob): %w", err)
 		}
@@ -324,6 +327,7 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) {
 		registry.metrics.Gauge("redemption_last_success_timestamp_seconds", nil).Set(0)
 		registry.metrics.Gauge("redemption_worker_last_completed_timestamp_seconds", nil).Set(0)
 		readiness.account = redeemerKey.Address()
+		readiness.observer = observedChain.Observer
 		redeemerDB.readiness = readiness
 	}
 	mainListener, err := net.Listen("tcp", p.Listen.String())
@@ -385,8 +389,9 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) {
 	startWorker(p.HeartbeatInterval, time.Second, func(context.Context) { registry.Expire() })
 	workers.Go(func() { app.runO2RKeepalives(ctx, 10*time.Second) })
 	if engine != nil {
+		workers.Go(func() { observedChain.Observer.Run(ctx) })
 		startWorker(p.HeartbeatInterval, 30*time.Second, func(ctx context.Context) {
-			readiness.check(ctx, redeemerDB, paymentChain, engine, registry.metrics, logger)
+			readiness.check(ctx, redeemerDB, observedChain, engine, registry.metrics, logger)
 		})
 		if p.ETHUSDFeed != nil {
 			startWorker(time.Hour, 30*time.Second, func(ctx context.Context) {
diff --git a/orchestrator/app_test.go b/orchestrator/app_test.go
index 2dc826e..6a6f3b9 100644
--- a/orchestrator/app_test.go
+++ b/orchestrator/app_test.go
@@ -95,7 +95,7 @@ func TestPaidHTTPStartup(t *testing.T) {
 					_ = json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": req.ID, "error": map[string]any{"code": -32000, "message": "ticket estimation unavailable"}})
 					return
 				case "eth_getBlockByNumber":
-					result = map[string]string{"number": "0x32", "hash": ethcommon.HexToHash("0xabcd").Hex()}
+					result = map[string]string{"number": "0x32", "hash": ethcommon.HexToHash("0xabcd").Hex(), "timestamp": fmt.Sprintf("0x%x", time.Now().Unix())}
 				case "eth_call":
 					// Reuse the 32-byte encoding of 1 for calls returning an address, round, hash, or bool.
 					result = fmt.Sprintf("0x%064x", 1)
@@ -164,8 +164,17 @@ func TestPaidHTTPStartup(t *testing.T) {
 			request, err := http.NewRequest(http.MethodPost, address+"/external/apps/r/session", nil)
 			require.NoError(t, err)
 			request.Header.Set("Livepeer-Payer-Address", payer.Address().Hex())
-			response, err = http.DefaultClient.Do(request)
-			require.NoError(t, err)
+			require.Eventually(t, func() bool {
+				response, err = http.DefaultClient.Do(request)
+				if err != nil {
+					return false
+				}
+				if response.StatusCode == http.StatusPaymentRequired {
+					return true
+				}
+				response.Body.Close()
+				return false
+			}, 3*time.Second, 10*time.Millisecond)
 			defer response.Body.Close()
 			require.Equal(t, http.StatusPaymentRequired, response.StatusCode)
 			var challenge Challenge
diff --git a/orchestrator/metrics.go b/orchestrator/metrics.go
index 8dda380..566421a 100644
--- a/orchestrator/metrics.go
+++ b/orchestrator/metrics.go
@@ -125,12 +125,16 @@ type paidReadiness struct {
 	chain, active, storage bool
 	snapshot               eth.ChainSnapshot
 	interval               time.Duration
+	observer               *eth.PaymentObserver
 }
 
 func (p *paidReadiness) ready(registry *Registry) bool {
 	p.mu.Lock()
 	limit := max(time.Minute, 3*p.interval)
 	ok := p.chain && p.active && p.storage && !p.completed.IsZero() && time.Since(p.completed) <= limit && !p.workerCompleted.IsZero() && time.Since(p.workerCompleted) <= limit
+	if p.observer != nil {
+		ok = ok && p.observer.Status().Healthy
+	}
 	p.mu.Unlock()
 	registry.mu.Lock()
 	fresh := registry.rateUntil.IsZero() || time.Now().Before(registry.rateUntil)
@@ -139,14 +143,14 @@ func (p *paidReadiness) ready(registry *Registry) bool {
 }
 
 func (p *paidReadiness) check(ctx context.Context, store *RedeemerDB, chain eth.PaymentChain, engine *PaymentEngine, metrics *observability.Metrics, logger *slog.Logger) {
+	dbCtx, cancel := context.WithTimeout(ctx, time.Second)
+	storageErr := store.checkStorage(dbCtx)
+	cancel()
 	snapshot, chainErr := chain.Snapshot(ctx)
 	active := false
 	if chainErr == nil {
 		active, chainErr = chain.IsActiveAt(ctx, engine.recipient, snapshot)
 	}
-	dbCtx, cancel := context.WithTimeout(ctx, time.Second)
-	storageErr := store.checkStorage(dbCtx)
-	cancel()
 	p.mu.Lock()
 	changed := p.completed.IsZero() || p.chain != (chainErr == nil) || p.active != active || p.storage != (storageErr == nil)
 	p.completed, p.chain, p.active, p.storage = time.Now(), chainErr == nil, active, storageErr == nil
@@ -163,7 +167,7 @@ func (p *paidReadiness) check(ctx context.Context, store *RedeemerDB, chain eth.
 		if snapshot.Round != nil {
 			metrics.Gauge("chain_snapshot_round", nil).Set(float64(snapshot.Round.Int64()))
 		}
-		metrics.Gauge("chain_snapshot_last_success_timestamp_seconds", nil).Set(float64(time.Now().Unix()))
+		metrics.Gauge("chain_snapshot_last_success_timestamp_seconds", nil).Set(float64(snapshot.ObservedAt().Unix()))
 	}
 	if balance, err := chain.Contracts.RPC.BalanceAt(ctx, p.account, nil); err == nil {
 		metrics.Gauge("redeemer_gas_balance_eth", nil).Set(observability.ETH(new(big.Rat).SetInt(balance)))
diff --git a/orchestrator/metrics_test.go b/orchestrator/metrics_test.go
index bf71073..744a232 100644
--- a/orchestrator/metrics_test.go
+++ b/orchestrator/metrics_test.go
@@ -3,6 +3,7 @@ package orchestrator
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"encoding/base64"
 	"encoding/json"
 	"io"
@@ -16,13 +17,85 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/livepeer/node/eth"
 	"github.com/livepeer/node/internal/observability"
 	"github.com/livepeer/node/internal/test"
+	"github.com/livepeer/node/pm"
 	"github.com/livepeer/node/pm/wire"
 	"github.com/stretchr/testify/require"
 )
 
+type snapshotReader struct {
+	pm.PaymentChain
+	read func(context.Context) (pm.ChainSnapshot, error)
+}
+
+func (c snapshotReader) Snapshot(ctx context.Context) (pm.ChainSnapshot, error) { return c.read(ctx) }
+
+func TestPaidReadinessUsesCurrentPublication(t *testing.T) {
+	f := test.NewChainRPC(t)
+	rpc, err := eth.OpenRPC(f.URL)
+	require.NoError(t, err)
+	defer rpc.Close()
+	contracts, err := eth.NewContracts(rpc, f.Controller)
+	require.NoError(t, err)
+	o := eth.NewPaymentObserver(eth.PaymentChain{Contracts: contracts}, true)
+	ctx, cancel := context.WithCancel(t.Context())
+	done := make(chan struct{})
+	go func() { o.Run(ctx); close(done) }()
+	defer func() { cancel(); <-done }()
+	var snapshot eth.ChainSnapshot
+	require.Eventually(t, func() bool { snapshot, err = o.Snapshot(ctx); return err == nil }, time.Second, time.Millisecond)
+	p := &paidReadiness{chain: true, active: true, storage: true, completed: time.Now(), workerCompleted: time.Now(), observer: o, snapshot: snapshot}
+	registry := NewRegistry("", "http://localhost", time.Second, time.Minute)
+	f.Update(func(s *test.ChainState) {
+		s.L1Block++
+		s.Hash = common.HexToHash("0xeeee")
+		s.Timestamp = uint64(time.Now().Unix())
+	})
+	require.Eventually(t, func() bool { return o.Status().ObservedAt.After(snapshot.ObservedAt()) }, 7*time.Second, 10*time.Millisecond)
+	require.ErrorIs(t, snapshot.Validate(), eth.ErrObservationUnavailable)
+	require.True(t, p.ready(registry), "an older heartbeat must not conceal a healthy completed backfill")
+	_, store, ticket, chain, _ := newRecoveryTest(t, 15)
+	store.readiness = p
+	stats, err := store.redemptionStats(t.Context())
+	require.NoError(t, err)
+	require.Equal(t, 1, stats.tickets["unknown"], "a healthy current observer cannot authorize a saved orphaned queue clock")
+	ticket.Sig = []byte{99, 2, 3}
+	require.ErrorIs(t, store.storeWinningTickets(t.Context(), ticket.PayerAddress, ticket.RecipientRandHash, 0, big.NewInt(10000), []*pm.SignedTicket{ticket}, snapshot), eth.ErrObservationUnavailable)
+	count, err := store.WinningTicketCount(ticket.PayerAddress, 0)
+	require.NoError(t, err)
+	require.Equal(t, 1, count, "invalidated admission must roll back its ticket insert")
+	captured := make(chan struct{})
+	e := &PaymentEngine{lastSeenBlock: new(big.Int).Set(snapshot.Block), nonceCount: 1, ticketNonces: map[string]*recipientNonces{"r": {nonceSeen: map[uint32]bool{1: true}, expirationBlock: new(big.Int).Add(snapshot.Block, big.NewInt(1))}}}
+	e.chain = snapshotReader{read: func(ctx context.Context) (pm.ChainSnapshot, error) {
+		s, err := o.Snapshot(ctx)
+		close(captured)
+		return s, err
+	}}
+	e.mu.Lock()
+	pruned := make(chan error, 1)
+	go func() { pruned <- e.PruneControlState(t.Context()) }()
+	select {
+	case <-captured:
+	case <-time.After(time.Second):
+		e.mu.Unlock()
+		t.Fatal("pruner did not capture snapshot")
+	}
+	cancel()
+	<-done
+	e.mu.Unlock()
+	require.ErrorIs(t, <-pruned, eth.ErrObservationUnavailable)
+	require.Equal(t, snapshot.Block, e.lastSeenBlock)
+	require.Len(t, e.ticketNonces, 1)
+	require.Equal(t, 1, e.nonceCount)
+	chain.Observer = o
+	p.check(t.Context(), store, chain, &PaymentEngine{recipient: ticket.Recipient}, observability.New("orchestrator"), slog.New(slog.NewTextHandler(io.Discard, nil)))
+	require.True(t, p.storage, "chain unavailability must not mark healthy storage unavailable")
+	require.False(t, p.ready(registry))
+}
+
 func TestRunnerCapacityOutcomes(t *testing.T) {
 	for _, paid := range []bool{false, true} {
 		name := "unpaid"
diff --git a/orchestrator/payment_engine.go b/orchestrator/payment_engine.go
index e60d6d7..ef528cf 100644
--- a/orchestrator/payment_engine.go
+++ b/orchestrator/payment_engine.go
@@ -110,6 +110,9 @@ func (e *PaymentEngine) makeChallenge(ctx context.Context, runner, manifest stri
 		Auth:         e.authToken(manifest, time.Now().Add(time.Hour).Unix())}
 	e.mu.Lock()
 	defer e.mu.Unlock()
+	if err := snapshot.Validate(); err != nil {
+		return Challenge{}, err
+	}
 	e.observeBlock(snapshot.Block)
 	current := e.sessions[manifest]
 	if current == nil {
@@ -262,6 +265,9 @@ func (e *PaymentEngine) Receive(ctx context.Context, runner, manifest, paymentHe
 	if e.sessions[manifest] != session {
 		return ethcommon.Address{}, nil, ErrMissingChallenge
 	}
+	if err := snapshot.Validate(); err != nil {
+		return ethcommon.Address{}, nil, paymentObservation{err, "error"}
+	}
 	e.observeBlock(snapshot.Block)
 	if params.ExpirationBlock.Cmp(e.lastSeenBlock) <= 0 {
 		return ethcommon.Address{}, nil, ErrInvalidPayment
@@ -292,7 +298,7 @@ func (e *PaymentEngine) Receive(ctx context.Context, runner, manifest, paymentHe
 			winners = append(winners, &pm.SignedTicket{Ticket: ticket, Sig: sp.Sig, RecipientRand: recipientRand})
 		}
 	}
-	if err := e.store.storeWinningTickets(ctx, payer, params.RecipientRandHash, snapshot.Round.Int64()-2, new(big.Int).Add(funds.Deposit, funds.Reserve), winners); err != nil {
+	if err := e.store.storeWinningTickets(ctx, payer, params.RecipientRandHash, snapshot.Round.Int64()-2, new(big.Int).Add(funds.Deposit, funds.Reserve), winners, snapshot); err != nil {
 		return ethcommon.Address{}, nil, err
 	}
 	if previous == nil {
diff --git a/orchestrator/payment_retention.go b/orchestrator/payment_retention.go
index 5368a9c..ff4dcf7 100644
--- a/orchestrator/payment_retention.go
+++ b/orchestrator/payment_retention.go
@@ -48,6 +48,9 @@ func (e *PaymentEngine) PruneControlState(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
+	if err := snapshot.Validate(); err != nil {
+		return err
+	}
 	if snapshot.Block != nil {
 		e.observeBlock(snapshot.Block)
 	}
diff --git a/orchestrator/redeemer_db.go b/orchestrator/redeemer_db.go
index c20f425..5ffb1b0 100644
--- a/orchestrator/redeemer_db.go
+++ b/orchestrator/redeemer_db.go
@@ -129,7 +129,7 @@ func (s *RedeemerDB) StoreWinningTicket(t *pm.SignedTicket) error {
 // storeWinningTickets commits a validated receipt batch before the payment
 // engine publishes replay guards or credit. Check exposure and liability in the
 // same transaction so redemption and concurrent receipts cannot race these gates.
-func (s *RedeemerDB) storeWinningTickets(ctx context.Context, payer ethcommon.Address, randHash ethcommon.Hash, minRound int64, collateral *big.Int, tickets []*pm.SignedTicket) error {
+func (s *RedeemerDB) storeWinningTickets(ctx context.Context, payer ethcommon.Address, randHash ethcommon.Hash, minRound int64, collateral *big.Int, tickets []*pm.SignedTicket, snapshot pm.ChainSnapshot) error {
 	tx, err := s.db.BeginTx(ctx, nil)
 	if err != nil {
 		return err
@@ -176,6 +176,9 @@ func (s *RedeemerDB) storeWinningTickets(ctx context.Context, payer ethcommon.Ad
 	if pending.Cmp(collateral) > 0 {
 		return pm.ErrPayerUnavailable
 	}
+	if err := snapshot.Validate(); err != nil {
+		return paymentObservation{err, "error"}
+	}
 	return tx.Commit()
 }
 
diff --git a/orchestrator/redemption_metrics.go b/orchestrator/redemption_metrics.go
index 2a1800b..457c914 100644
--- a/orchestrator/redemption_metrics.go
+++ b/orchestrator/redemption_metrics.go
@@ -73,7 +73,7 @@ func (s *RedeemerDB) redemptionStats(ctx context.Context) (stats redemptionStats
 	var block, round *big.Int
 	if s.readiness != nil {
 		s.readiness.mu.Lock()
-		if time.Since(s.readiness.completed) <= max(time.Minute, 3*s.readiness.interval) && s.readiness.chain {
+		if time.Since(s.readiness.completed) <= max(time.Minute, 3*s.readiness.interval) && s.readiness.chain && s.readiness.snapshot.Validate() == nil {
 			block, round = s.readiness.snapshot.Block, s.readiness.snapshot.Round
 		}
 		s.readiness.mu.Unlock()
diff --git a/orchestrator/redemption_nonce_test.go b/orchestrator/redemption_nonce_test.go
index 3f17079..4c9a047 100644
--- a/orchestrator/redemption_nonce_test.go
+++ b/orchestrator/redemption_nonce_test.go
@@ -134,7 +134,7 @@ func TestFinalizedExternalNonceRetiresOnlyMissingTransactions(t *testing.T) {
 			require.Empty(t, ReconcileSubmitted(t.Context(), s, c))
 			next := nextRecoveryTicket(ticket, 235)
 			next.CreationRound = 5
-			require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 5, big.NewInt(1000), []*pm.SignedTicket{next}), pm.ErrPayerUnavailable)
+			require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 5, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), pm.ErrPayerUnavailable)
 			f.mu.Lock()
 			f.finalizedNonce = 1
 			f.finalizedNonceReads = 0
@@ -153,10 +153,10 @@ func TestFinalizedExternalNonceRetiresOnlyMissingTransactions(t *testing.T) {
 			if observation == "missing" {
 				require.Empty(t, items)
 				require.Equal(t, "expired", ticketPhase(t, s, ticket.Sig))
-				require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 5, big.NewInt(1000), []*pm.SignedTicket{next}))
+				require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 5, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}))
 				oldEpoch := nextRecoveryTicket(ticket, 236)
 				oldEpoch.RecipientRandHash = ticket.RecipientRandHash
-				require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, ticket.RecipientRandHash, 0, big.NewInt(10000), []*pm.SignedTicket{oldEpoch}), ErrInvalidPayment)
+				require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, ticket.RecipientRandHash, 0, big.NewInt(10000), []*pm.SignedTicket{oldEpoch}, pm.ChainSnapshot{}), ErrInvalidPayment)
 			} else {
 				require.Len(t, items, 1, "uncertain or unfinalized receipts retain liability")
 			}
@@ -191,7 +191,7 @@ func TestFinalizedExternalNoncePreservesHigherAttempt(t *testing.T) {
 	require.Equal(t, 1, reads)
 	next := nextRecoveryTicket(ticket, 238)
 	next.CreationRound = 5
-	require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 5, big.NewInt(1000), []*pm.SignedTicket{next}), pm.ErrPayerUnavailable)
+	require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 5, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), pm.ErrPayerUnavailable)
 	// Finalized state advances beyond both attempts while the pending endpoint
 	// stays stale. New signing must use that observed count, not the saved floor.
 	f.mu.Lock()
diff --git a/orchestrator/redemption_recovery_test.go b/orchestrator/redemption_recovery_test.go
index 7db0a03..8e2f7ec 100644
--- a/orchestrator/redemption_recovery_test.go
+++ b/orchestrator/redemption_recovery_test.go
@@ -694,11 +694,11 @@ func TestTerminalAttemptRetainsUnresolvedSibling(t *testing.T) {
 			// stops retrying, but cannot release its collateral based on one failed hash.
 			next := nextRecoveryTicket(ticket, 12)
 			next.CreationRound = max(ticket.CreationRound, tc.minRound)
-			require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, tc.minRound, big.NewInt(1000), []*pm.SignedTicket{next}), pm.ErrPayerUnavailable)
+			require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, tc.minRound, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), pm.ErrPayerUnavailable)
 			f.mine(sent[0], true)
 			tickRecovery(t, s, c, key)
 			require.Equal(t, "confirmed", ticketPhase(t, s, ticket.Sig))
-			require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, tc.minRound, big.NewInt(1000), []*pm.SignedTicket{next}))
+			require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, tc.minRound, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}))
 		})
 	}
 }
@@ -771,7 +771,7 @@ func TestExpiredAttemptDoesNotReopenRandomness(t *testing.T) {
 	next := nextRecoveryTicket(ticket, 17)
 	next.RecipientRandHash = ticket.RecipientRandHash
 	next.TicketNonce = 2
-	require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, ticket.RecipientRandHash, 0, big.NewInt(10000), []*pm.SignedTicket{next}), ErrInvalidPayment)
+	require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, ticket.RecipientRandHash, 0, big.NewInt(10000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), ErrInvalidPayment)
 }
 
 func recoveryReplacement(t *testing.T, key *eth.Key, first *types.Transaction) *types.Transaction {
@@ -870,7 +870,7 @@ func TestRevertedReplacementReleasesSameNonceLiability(t *testing.T) {
 	require.NotEmpty(t, ReconcileSubmitted(t.Context(), s, c))
 	require.Equal(t, "reverted", ticketPhase(t, s, ticket.Sig))
 	next := nextRecoveryTicket(ticket, 21)
-	require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}), "a finalized replacement consumes every sibling with the same sender and nonce")
+	require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), "a finalized replacement consumes every sibling with the same sender and nonce")
 }
 
 func TestManualRetryFailureKeepsActiveMonitor(t *testing.T) {
@@ -942,7 +942,7 @@ func TestConsumedNonceResolvesOtherTicketHistory(t *testing.T) {
 	require.EqualValues(t, 1, test.MetricSum(t, s.metrics.Registry, "redemption_settlements_total", nil))
 	require.NoError(t, s.setTicketPhase(t.Context(), sibling.Sig, "invalid"))
 	next := nextRecoveryTicket(a, 105)
-	require.NoError(t, s.storeWinningTickets(t.Context(), a.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}), "B consumed nonce zero and A's nonce one failed; A has no remaining liability")
+	require.NoError(t, s.storeWinningTickets(t.Context(), a.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), "B consumed nonce zero and A's nonce one failed; A has no remaining liability")
 }
 
 func TestMultipleFinalizedFailuresResolveEveryGroup(t *testing.T) {
@@ -973,7 +973,7 @@ func TestMultipleFinalizedFailuresResolveEveryGroup(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, history, 4)
 	next := nextRecoveryTicket(ticket, 107)
-	require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}), "both finalized failed nonces must resolve their replacement siblings")
+	require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), "both finalized failed nonces must resolve their replacement siblings")
 }
 
 func TestLiveAcknowledgementRestoresNonce(t *testing.T) {
@@ -1038,7 +1038,7 @@ func TestUsedTransitionRecoversAfterStorageFailure(t *testing.T) {
 		t.Errorf("summary=%s but redeemed_at remains NULL even after restart and retry", ticketPhase(t, s, ticket.Sig))
 	}
 	next := nextRecoveryTicket(ticket, 223)
-	require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}), "a finalized-used ticket must not strand collateral")
+	require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), "a finalized-used ticket must not strand collateral")
 }
 
 func TestRejectedReplacementKeepsPredecessorReconcilable(t *testing.T) {
diff --git a/signer/app.go b/signer/app.go
index 6f9edc1..9ce2a96 100644
--- a/signer/app.go
+++ b/signer/app.go
@@ -183,7 +183,10 @@ func serve(parent context.Context, p Params) error {
 	if err != nil {
 		return fmt.Errorf("initialize signer contract bindings: %w", err)
 	}
-	service.SetPaymentChain(pm.EthereumChain{Client: eth.PaymentChain{Contracts: contracts}})
+	chain := eth.PaymentChain{Contracts: contracts}
+	observer := eth.NewPaymentObserver(chain, p.Network == nodeconfig.Mainnet)
+	chain.Observer = observer
+	service.SetPaymentChain(pm.EthereumChain{Client: chain})
 	policy, _ := newPricePolicy(p.MaxHourlyPrice, p.MaxFixedPrice)
 	policy.metrics = service.metrics
 	if p.WeiPerUSD == nil {
@@ -253,6 +256,7 @@ func serve(parent context.Context, p Params) error {
 	metricsServer := &http.Server{BaseContext: srv.BaseContext, Handler: metricsMux, ReadHeaderTimeout: 5 * time.Second, WriteTimeout: 10 * time.Second, IdleTimeout: 30 * time.Second, MaxHeaderBytes: 16 << 10, MaxHeaderValueCount: 128}
 	errCh := make(chan error, 2)
 	var workers sync.WaitGroup
+	workers.Go(func() { observer.Run(ctx) })
 	// Keep delivery alive until in-flight HTTP handlers have finished enqueuing.
 	publisherCtx, publisherStop := context.WithCancel(context.Background())
 	defer publisherStop()
diff --git a/signer/app_test.go b/signer/app_test.go
index 09d1796..8904cec 100644
--- a/signer/app_test.go
+++ b/signer/app_test.go
@@ -297,9 +297,13 @@ func TestSignerReadiness(t *testing.T) {
 					result = "0x1"
 				case "eth_getBlockByNumber":
 					if failure.Load() != 1 {
-						result = map[string]string{"number": "0x32", "hash": ethcommon.HexToHash("0x1234").Hex()}
+						result = map[string]string{"number": "0x32", "hash": ethcommon.HexToHash("0x1234").Hex(), "timestamp": fmt.Sprintf("0x%x", time.Now().Unix())}
 					}
 				case "eth_call":
+					if failure.Load() == 1 {
+						_ = json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "error": map[string]any{"code": -32000}})
+						return
+					}
 					var call struct {
 						Data string `json:"input"`
 					}
@@ -310,7 +314,7 @@ func TestSignerReadiness(t *testing.T) {
 					switch {
 					case selector("getContract(bytes32)"):
 						result = fmt.Sprintf("0x%064x", 1)
-					case selector("lastInitializedRound()"):
+					case selector("lastInitializedRound()"), selector("currentRound()"):
 						result = fmt.Sprintf("0x%064x", 5)
 					case selector("blockHashForRound(uint256)"):
 						result = ethcommon.HexToHash("0x1234").Hex()
diff --git a/signer/metrics.go b/signer/metrics.go
index a7983b3..c7034ec 100644
--- a/signer/metrics.go
+++ b/signer/metrics.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"math/big"
-	"time"
 
 	"github.com/livepeer/node/internal/observability"
 	"github.com/livepeer/node/pm"
@@ -56,7 +55,7 @@ func (s *Service) signMessage(kind string, data []byte) ([]byte, error) {
 
 func (s *Service) observeFunds(funds pm.PayerFunds, total bool) {
 	if funds.Snapshot.Block != nil && funds.Snapshot.Round != nil {
-		s.metrics.Gauge("chain_snapshot_last_success_timestamp_seconds", nil).Set(float64(time.Now().Unix()))
+		s.metrics.Gauge("chain_snapshot_last_success_timestamp_seconds", nil).Set(float64(funds.Snapshot.ObservedAt().Unix()))
 	}
 	if funds.Snapshot.Block != nil {
 		s.metrics.Gauge("chain_snapshot_block", nil).Set(float64(funds.Snapshot.Block.Int64()))
diff --git a/signer/payment.go b/signer/payment.go
index c547e1f..b7735a8 100644
--- a/signer/payment.go
+++ b/signer/payment.go
@@ -22,11 +22,12 @@ const paramsExpiryBuffer = int64(1)
 // draftPayment fixes the payment and accounting inputs before authorization.
 // Only State's authorization fields are updated by authorizePayment.
 type draftPayment struct {
-	batch   *pm.TicketBatch
-	payment wire.Payment
-	segment wire.SegData
-	State   paymentState
-	Usage   paymentUsage
+	batch    *pm.TicketBatch
+	payment  wire.Payment
+	segment  wire.SegData
+	State    paymentState
+	Usage    paymentUsage
+	snapshot pm.ChainSnapshot
 }
 
 type signedPayment struct {
@@ -150,13 +151,16 @@ func (s *Service) preparePayment(ctx context.Context, paymentType, manifest stri
 	state.Balance = remaining.RatString()
 	state.LastUpdate = now
 	state.SequenceNumber = uint64(oldSequence + 1)
-	return draftPayment{batch: batch, payment: message, segment: segment, State: state,
+	return draftPayment{batch: batch, payment: message, segment: segment, State: state, snapshot: funds.Snapshot,
 		Usage: paymentUsage{Fee: fee, Balance: remaining, PreviousTime: previous, BillableSecs: billableSecs, NumTickets: count}}, nil
 }
 
 // Sign adds signatures to the exact payment prepared before authorization.
-// It does not observe the clock or chain, or recalculate any payment values.
+// It validates the retained observation locally without another chain read.
 func (p draftPayment) Sign(ctx context.Context, signer pm.TicketSigner, metrics *observability.Metrics) (signedPayment, error) {
+	if err := p.snapshot.Validate(); err != nil {
+		return signedPayment{}, errors.Join(errPaymentService, err)
+	}
 	message := p.payment
 	message.PayerParams = make([]wire.TicketPayerParams, 0, len(p.batch.PayerParams))
 	for _, ticket := range p.batch.Tickets() {
@@ -178,6 +182,9 @@ func (p draftPayment) Sign(ctx context.Context, signer pm.TicketSigner, metrics
 	}
 	segment := p.segment
 	segment.Signature = sig
+	if err := p.snapshot.Validate(); err != nil {
+		return signedPayment{}, errors.Join(errPaymentService, err)
+	}
 	return signedPayment{Payment: base64.StdEncoding.EncodeToString(wire.EncodePayment(message)), SegCreds: base64.StdEncoding.EncodeToString(wire.EncodeSegData(segment))}, nil
 }
 
diff --git a/signer/payment_test.go b/signer/payment_test.go
index a55e73e..9317339 100644
--- a/signer/payment_test.go
+++ b/signer/payment_test.go
@@ -200,4 +200,56 @@ func TestPaidRequestDirectRPCBudget(t *testing.T) {
 		"RoundsManager.lastInitializedRound": 3, "RoundsManager.blockHashForRound": 3,
 		"BondingManager.isActiveTranscoder": 2, "TicketBroker.getSenderInfo": 2, "TicketBroker.claimableReserve": 2,
 	}, stats.Calls)
+
+	// A signer's retained observation cannot prove that the recipient is behind
+	// expiry. The recipient must still enforce its own clock at the boundary.
+	_, err = engine.MakeChallenge(t.Context(), "runner", "divergent", s.key.Address(), 9, "fixed", "https://orch.example")
+	require.NoError(t, err)
+	info, err = engine.ChallengeInfo("divergent")
+	require.NoError(t, err)
+	older, err := direct.PayerFunds(t.Context(), s.key.Address(), recipient)
+	require.NoError(t, err)
+	s.SetPaymentChain(payerChainFunc(func(context.Context, ethcommon.Address, ethcommon.Address) (pm.PayerFunds, error) { return older, nil }))
+	rpc.Update(func(state *test.ChainState) {
+		state.L1Block = new(big.Int).SetBytes(info.TicketParams.ExpirationBlock).Uint64()
+		state.Number++
+		state.Hash = ethcommon.HexToHash("0xeeee")
+	})
+	response = postPayment(t, s, map[string]any{"orchestrator": wire.EncodeOrchestratorInfo(info), "type": "fixed"})
+	require.Equal(t, 200, response.Code, response.Body.String())
+	require.NoError(t, json.Unmarshal(response.Body.Bytes(), &signed))
+	_, _, err = engine.Receive(t.Context(), "runner", "divergent", signed.Payment, signed.SegCreds)
+	require.ErrorIs(t, err, orchestrator.ErrInvalidPayment)
+
+	const manifest = "invalidated-signing"
+	_, err = engine.MakeChallenge(t.Context(), "runner", manifest, s.key.Address(), 9, "fixed", "https://orch.example")
+	require.NoError(t, err)
+	info, err = engine.ChallengeInfo(manifest)
+	require.NoError(t, err)
+	for _, during := range []bool{false, true} {
+		observer := eth.NewPaymentObserver(direct.Client, false)
+		ctx, cancel := context.WithCancel(t.Context())
+		done := make(chan struct{})
+		go func() { observer.Run(ctx); close(done) }()
+		stop := func() { cancel(); <-done }
+		defer stop()
+		require.Eventually(t, func() bool { _, err := observer.Snapshot(ctx); return err == nil }, time.Second, time.Millisecond)
+		observed := direct
+		observed.Client.Observer = observer
+		s.SetPaymentChain(observed)
+		draft, err := s.preparePayment(t.Context(), "fixed", manifest, info, paymentState{}, -1, time.Now())
+		require.NoError(t, err)
+		key := &failingPaymentSigner{TicketSigner: s.key}
+		if during {
+			key.before = stop
+		} else {
+			stop()
+		}
+		payment, err := draft.Sign(t.Context(), key, s.metrics)
+		require.ErrorIs(t, err, eth.ErrObservationUnavailable)
+		require.Empty(t, payment)
+		if !during {
+			require.Zero(t, key.calls.Load(), "invalidated observations must not produce signatures")
+		}
+	}
 }
diff --git a/signer/service_test.go b/signer/service_test.go
index ceef660..6cb07ee 100644
--- a/signer/service_test.go
+++ b/signer/service_test.go
@@ -410,9 +410,14 @@ type failingPaymentSigner struct {
 	pm.TicketSigner
 	failAt int
 	calls  atomic.Int32
+	before func()
 }
 
 func (s *failingPaymentSigner) SignMessage(message []byte) ([]byte, error) {
+	if s.before != nil {
+		s.before()
+		s.before = nil
+	}
 	if int(s.calls.Add(1)) == s.failAt {
 		return nil, errors.New("key /private/keys/signer.json: private-signing-detail")
 	}