consensus/engine/chain/engine.go

package chain

import (
	"context"
	"errors"
	"sync"
	"time"

	"github.com/luxfi/consensus/config"
	"github.com/luxfi/consensus/core/slashing"
	"github.com/luxfi/consensus/engine"
	"github.com/luxfi/consensus/engine/chain/block"
	"github.com/luxfi/ids"
	"github.com/luxfi/log"
)

// -----------------------------------------------------------------------------
// Errors
// -----------------------------------------------------------------------------

var (
	ErrNotStarted     = errors.New("engine not started")
	ErrAlreadyStarted = errors.New("engine already started")

	// ErrQuorumVerifierRequired is returned by Start when a multi-validator
	// engine (K>1) is started without a VoteVerifier. Multi-validator finality
	// MUST be gated on a verifiable α-of-K quorum cert; without a verifier
	// there is no way to tell a real quorum from forged votes. Fail-closed.
	ErrQuorumVerifierRequired = errors.New("chain: multi-validator engine (K>1) requires a vote verifier for quorum-cert finality (use WithQuorumCert / WithVoteVerifier)")
)

// -----------------------------------------------------------------------------
// Interfaces
// -----------------------------------------------------------------------------

// Engine is the chain consensus engine interface.
type Engine interface {
	Start(context.Context, bool) error
	StopWithError(context.Context, error) error
	Context() context.Context
	HealthCheck(context.Context) (interface{}, error)
	IsBootstrapped() bool
}

// BlockBuilder is the minimal interface consensus needs from a VM.
// Intentionally small: VMs may implement more, but consensus only needs these.
type BlockBuilder interface {
	BuildBlock(context.Context) (block.Block, error)
	GetBlock(context.Context, ids.ID) (block.Block, error)
	ParseBlock(context.Context, []byte) (block.Block, error)
	LastAccepted(context.Context) (ids.ID, error)
	// SetPreference tells the VM which block to build on next.
	// This MUST be called after accepting a block to keep the VM's preferred
	// block in sync with the last accepted block. Without this, the VM's
	// Preferred() returns the old block while GetLastAccepted() returns the
	// new block, causing GetState(preferred) to fail during block building.
	SetPreference(context.Context, ids.ID) error
}

// BlockProposer submits blocks to validators.
// Consensus expresses WHAT (propose block); implementation decides HOW.
type BlockProposer interface {
	Propose(ctx context.Context, proposal BlockProposal) error
	RequestVotes(ctx context.Context, req VoteRequest) error
}

// VoteEmitter is deprecated. Use BlockProposer.
type VoteEmitter = BlockProposer

// -----------------------------------------------------------------------------
// Message types
// -----------------------------------------------------------------------------

type (
	MessageType = engine.MessageType
	Message     = engine.Message
)

const (
	PendingTxs    = engine.PendingTxs
	StateSyncDone = engine.StateSyncDone
)

// -----------------------------------------------------------------------------
// Protocol types
// -----------------------------------------------------------------------------

// BlockProposal contains data needed to propose a block.
type BlockProposal struct {
	BlockID   ids.ID
	BlockData []byte
	Height    uint64
	ParentID  ids.ID
}

// VoteRequest asks specific validators to vote.
type VoteRequest struct {
	BlockID    ids.ID
	BlockData  []byte // Block bytes for PushQuery (peers can immediately verify and vote)
	Validators []ids.NodeID
}

// Vote represents a validator's decision.
type Vote struct {
	BlockID  ids.ID
	NodeID   ids.NodeID
	Accept   bool
	SignedAt time.Time

	// Signature is the validator's signature over CanonicalVoteMessage of the
	// block's position (ChainID, Height, Round, BlockID, ParentID). It is the
	// material the engine collects into a QuorumCert — the portable, verifiable
	// α-of-K finality witness. May be empty for a single-validator (K==1)
	// engine, where the sole validator's local accept is the quorum and no cert
	// is gossiped; MUST be present and valid for multi-validator finality.
	Signature []byte
	// ParentID binds the parent into the vote's signed position. Carried so the
	// engine can rebuild CanonicalVoteMessage when assembling a cert even if it
	// is not separately tracking the block's parent.
	ParentID ids.ID
	// Round is the consensus round the vote was cast in (0 for the first round
	// at a height). Bound into the signed position.
	Round uint32
}

// PendingBlock tracks a block awaiting consensus.
type PendingBlock struct {
	ConsensusBlock *Block
	VMBlock        block.Block
	ProposedAt     time.Time
	VoteCount      int // Accept votes
	RejectCount    int // Reject votes
	Decided        bool

	// certVotes collects the distinct SIGNED accept votes observed for this
	// block, keyed by voter NodeID (de-dup: one vote per validator). When the
	// count reaches alpha the engine assembles a QuorumCert from these — the
	// α-of-K finality witness. Empty for single-validator (K==1) finality.
	certVotes map[ids.NodeID]SignedVote
	// Round is the consensus round the block was proposed in. Bound into every
	// vote's signed position so a cert binds the exact round.
	Round uint32
	// cert is the assembled+verified finality witness once the quorum is
	// reached (nil until then). Retained so the engine can re-gossip it on
	// request and so a follower's accept is gated on holding it.
	cert *QuorumCert

	// lastRePoll is when the re-poll loop last re-solicited votes for this block
	// (zero until the first re-poll). The re-poll loop re-drives a block at most
	// once per its CURRENT backoff window (rePollBackoff), so a stuck block
	// recovers without a gossip storm. See rePollAllPending — this is the liveness
	// retry the topology doc promises ("vote-broadcast + cert-gossip + the
	// poll-timeout re-request"), now with exponential backoff + a hard cap.
	lastRePoll time.Time

	// rePollBackoff is the CURRENT re-poll interval for this block. It starts at
	// the base RoundTO and DOUBLES after each re-poll (capped at maxRePollBackoff),
	// so a block that is stuck because peers are behind is re-solicited on a
	// geometric schedule (RoundTO, 2·RoundTO, 4·RoundTO, …), not a 250ms hot loop.
	// Zero ⇒ "use the base interval for the first re-poll".
	rePollBackoff time.Duration

	// rePollAttempts counts how many times the re-poll loop has re-solicited this
	// block. Once it reaches maxRePollAttempts the block is ABANDONED for re-poll
	// purposes (rePollAbandoned) — re-soliciting the SAME block to peers who
	// cannot vote (they are behind its parent) is pure spam and never recovers it;
	// the catch-up path (requestCatchup) is what recovers a behind frontier.
	rePollAttempts int

	// rePollAbandoned is set once rePollAttempts hits the cap. An abandoned block
	// is NEVER re-polled again (no infinite spam), but is NOT deleted: it remains
	// pending and recoverable — a late cert (HandleIncomingCert) or a catch-up
	// that fills its parent can still finalize it.
	rePollAbandoned bool

	// IsOwnProposal is true when this node built and proposed the block. It now
	// selects ONLY the finalization ENTRY POINT (finalizeOwnProposal vs.
	// tryFinalizeBlock); it does NOT alter how votes are counted. The former
	// REJECT→ACCEPT laundering it used to gate ("a peer's reject counts as accept
	// for my own block") is DELETED — vote.Accept is authoritative once the
	// signature verifies (handleVote), so an own-proposal finalizes only on a
	// real α-of-K cert (K>1) or the 1-of-1 force (K==1), never on self-promises.
	IsOwnProposal bool
}

// -----------------------------------------------------------------------------
// Configuration
// -----------------------------------------------------------------------------

// Config holds engine dependencies and settings.
type Config struct {
	Params   config.Parameters
	VM       BlockBuilder
	Proposer BlockProposer

	// Channel buffer sizes (defaults applied if zero)
	VoteRequestBuffer int
	VoteBuffer        int
}

// DefaultConfig returns sensible defaults.
func DefaultConfig() Config {
	return Config{
		Params:            config.DefaultParams(),
		VoteRequestBuffer: 100,
		VoteBuffer:        1000,
	}
}

// Validate checks config validity.
func (c Config) Validate() error {
	return c.Params.Valid()
}

// -----------------------------------------------------------------------------
// Functional options
// -----------------------------------------------------------------------------

// Option configures the engine.
type Option func(*Transitive)

// WithParams sets consensus parameters.
func WithParams(p config.Parameters) Option {
	return func(t *Transitive) {
		t.params = p
		t.consensus = NewChainConsensus(p.K, p.AlphaPreference, int(p.Beta))
	}
}

// WithVM sets the block builder.
func WithVM(vm BlockBuilder) Option {
	return func(t *Transitive) {
		t.vm = vm
	}
}

// WithProposer sets the block proposer.
func WithProposer(p BlockProposer) Option {
	return func(t *Transitive) {
		t.proposer = p
	}
}

// WithEmitter is deprecated. Use WithProposer.
func WithEmitter(e BlockProposer) Option {
	return WithProposer(e)
}

// WithSlashing enables equivocation detection and slashing evidence collection.
func WithSlashing(detector *slashing.Detector, db *slashing.DB) Option {
	return func(t *Transitive) {
		t.slashingDetector = detector
		t.slashingDB = db
	}
}

// WithLogger sets the engine logger.
func WithLogger(l log.Logger) Option {
	return func(t *Transitive) {
		t.log = l
	}
}

// WithQuorumCert wires multi-validator α-of-K cert-witnessed finality. The node
// supplies a VoteVerifier (mandatory for cert finality — verifies every vote
// signature and every incoming cert), and optionally a CertGossiper (to
// distribute assembled certs) and a VoteSigner (to sign this node's own votes).
// chainID and nodeID identify this node's position for vote/cert binding.
//
// Without this option the engine runs in single-validator (K==1) mode: the sole
// validator's local accept is the quorum, finality uses the ForceAccept path,
// and no certs or signatures are produced. The engine REFUSES to start a
// multi-validator (K>1) configuration without a verifier (fail-closed) — see
// Start.
func WithQuorumCert(chainID ids.ID, nodeID ids.NodeID, verifier VoteVerifier, gossiper CertGossiper, signer VoteSigner) Option {
	return func(t *Transitive) {
		t.chainID = chainID
		t.nodeID = nodeID
		t.voteVerifier = verifier
		t.certGossiper = gossiper
		t.voteSigner = signer
	}
}

// WithVoteVerifier sets only the vote/cert signature verifier. Convenience for
// callers that verify but neither sign nor gossip (e.g. a verifying-only node
// or a test).
func WithVoteVerifier(verifier VoteVerifier) Option {
	return func(t *Transitive) {
		t.voteVerifier = verifier
	}
}

// WithStakeWeighting makes finality STAKE-WEIGHTED (HIGH-3): a cert is accepted
// only when its voters hold a strict ⅔-of-stake supermajority at the cert's
// height, in addition to the α-of-K count. The node supplies a StakeSource
// backed by the chain's validator set (weights from the same set the verifier
// authenticates against). REQUIRED for a PoS value chain with unequal stake;
// omit it only when equal stake is enforced at admission (then count-α is the
// correct, equivalent rule). It is ALSO a precondition for Mode() to report
// ModeQuorumFinality (the engine's stake-weighted finality regime) — see Mode().
func WithStakeWeighting(stake StakeSource) Option {
	return func(t *Transitive) {
		t.stakeSource = stake
	}
}

// WithStrictPQ marks the engine as running under a STRICT post-quantum security
// profile (the node derives this from the chain's consensus profile —
// config.Profile.IsStrict()). When set, Mode() additionally requires a PQ
// cryptoWitness (WithCryptoWitness) before it reports ModeQuorumFinality, so the
// engine cannot report a quorum-finality regime the chain cannot witness
// post-quantum. A non-strict chain leaves this false (the requirement is vacuous).
func WithStrictPQ(strict bool) Option {
	return func(t *Transitive) {
		t.strictPQ = strict
	}
}

// WithCryptoWitness wires the post-quantum finality witness source a strict-PQ
// chain uses. It is REQUIRED for Mode() to report ModeQuorumFinality on a strict-PQ
// chain; on a non-strict chain it is unused. The node supplies a source whose
// Scheme() names the PQ witness scheme actually in force.
func WithCryptoWitness(w CryptoWitnessSource) Option {
	return func(t *Transitive) {
		t.cryptoWitness = w
	}
}

// WithCatchup wires the engine's runtime auto-recovery seam (Catchup): the
// transport the engine uses to fetch ancestors it is missing when a gossiped
// child or a verified cert references a parent it does not have. Without it a
// follower that falls behind during normal operation is stranded (it can neither
// vote on nor finalize a block whose parent it lacks); with it the follower
// self-heals by fetching the missing chain and rejoining the frontier. Optional;
// nil keeps the legacy no-self-heal behaviour.
func WithCatchup(c Catchup) Option {
	return func(t *Transitive) {
		t.catchup = c
	}
}

// WithValidatorSetRoot binds every vote/cert this engine produces to the active
// weighted validator set at the block's height (the MEDIUM fix). The node
// supplies a ValidatorSetRootSource backed by the chain's validator set; the
// engine stamps the root into each VotePosition so a cert is pinned to the exact
// set it was certified under — a cross-epoch cert (votes cast under set R
// re-presented as certifying under set R') fails signature verification because
// every signature was over R. REQUIRED alongside WithStakeWeighting on a chain
// whose validator set / stake can change across epochs, so the ⅔-by-stake
// predicate is enforced at the cert-position epoch rather than assumed. Omit it
// only on a fixed-set chain (then Empty-root binding is the correct no-op).
func WithValidatorSetRoot(src ValidatorSetRootSource) Option {
	return func(t *Transitive) {
		t.setRootSource = src
	}
}

// WithVoteBuffers sets channel buffer sizes.
func WithVoteBuffers(requests, votes int) Option {
	return func(t *Transitive) {
		if requests > 0 {
			t.voteRequests = make(chan VoteRequest, requests)
		}
		if votes > 0 {
			t.votes = make(chan Vote, votes)
		}
	}
}

// -----------------------------------------------------------------------------
// Transitive consensus engine
// -----------------------------------------------------------------------------

// Transitive implements chain consensus using Lux protocols.
//
// Construction:
//
//	New()                              // defaults
//	New(WithVM(vm), WithProposer(p))   // with options
//	NewWithConfig(cfg)                 // explicit config
//	NewWithConfig(cfg, WithVM(vm))     // config + option overrides
//
// Lifecycle: New -> Start -> (running) -> Stop
type Transitive struct {
	mu sync.RWMutex

	// Core consensus
	consensus *ChainConsensus
	params    config.Parameters

	// Dependencies
	vm       BlockBuilder
	proposer BlockProposer

	// Runtime state
	ctx          context.Context
	cancel       context.CancelFunc
	bootstrapped bool
	started      bool
	wg           sync.WaitGroup // tracks background goroutines

	// bootstrapPhase is true while the node is INITIAL-SYNCING — fetching and
	// executing the chain from a peer's accepted frontier down to its local tip —
	// and false once it has reached the frontier and entered live consensus. It is
	// the SOLE gate on the bootstrap accept authority (Runtime.AcceptBootstrapBlock):
	// a block fetched-from-frontier-and-re-executed may finalize WITHOUT an α-of-K
	// cert ONLY while this is true. The instant the node goes live (FinishBootstrap),
	// the bootstrap path is fail-closed and finality flows ONLY through the
	// cert-witnessed α-of-K road — so the weak-subjectivity-on-the-beacon-set trust
	// of bootstrap can never be used to bypass the live cert-gate. Defaults true at
	// construction (a fresh engine is bootstrapping); the node flips it false exactly
	// when it signals the chain bootstrapped.
	bootstrapPhase bool

	// Block management
	pendingBlocks      map[ids.ID]*PendingBlock
	pendingBuildBlocks int

	// finalizedByCert is the engine's authoritative finality record: the set of
	// block IDs that were committed through the SOLE cert-gated finalizer
	// (AcceptWithCert, which requires a VerifiedQuorumCert). It is deliberately
	// SEPARATE from the consensus core's block.accepted / finalizedByHeight, which
	// the α-of-K COUNT path populates directly (consensus.go marks a block accepted
	// on acceptVotes>=alpha). A block sitting at count-α but lacking the verified
	// cert is "accepted" in the consensus core but is NOT in this set — and
	// Transitive.IsAccepted reports THIS set, so the engine never claims finality
	// for a block it refused to finalize. Bounded by the same finalize cadence as
	// the chain (one entry per finalized height); a production node prunes it
	// alongside the slashing window if retention ever matters.
	finalizedByCert map[ids.ID]struct{}

	// certBytesByBlock persists the marshaled finality cert for each block this node
	// finalized, so it can SERVE that cert to a peer catching up (CertForBlock). It
	// is written at the SOLE finalizer (acceptWithCertCore), so every finalize path
	// captures its cert in ONE place. certServedOrder is the companion FIFO of block
	// ids in finalize (== ascending height) order, used to evict the oldest cert once
	// the store passes maxServedCerts — a bounded sliding window, never an unbounded
	// map. A node lagging beyond the window bootstraps instead of catching up.
	certBytesByBlock map[ids.ID][]byte
	certServedOrder  []ids.ID

	// Vote channels
	voteRequests chan VoteRequest
	votes        chan Vote

	// Pipeline: signal channel for continuous block production
	pipelineSignal chan struct{}

	// Metrics
	blocksBuilt    uint64
	blocksAccepted uint64
	blocksRejected uint64
	votesSent      uint64
	votesReceived  uint64

	// Slashing: equivocation detection (optional, nil disables)
	slashingDetector *slashing.Detector
	slashingDB       *slashing.DB

	// Quorum-cert finality (multi-validator). These are the engine's sole
	// dependencies for α-of-K cert-witnessed finality:
	//
	//   - voteVerifier verifies each collected vote's signature before it is
	//     counted toward a cert and verifies an incoming cert's signatures.
	//     The node injects a real scheme (BLS / ML-DSA / secp256k1). When nil,
	//     the engine is in single-validator (K==1) mode and finality uses the
	//     local-accept force path (no cert, no signatures).
	//   - certGossiper re-broadcasts an assembled cert to all validators so
	//     followers can finalize on a verifiable α-of-K proof rather than
	//     fast-following an unverified block. Optional (nil disables cert
	//     gossip; finality still holds locally via the α-of-K count).
	//   - voteSigner signs this node's own accept votes (used when it votes as
	//     a follower, so its signature can be collected into a cert). Optional.
	//   - chainID / nodeID identify this node's position for vote/cert binding.
	voteVerifier VoteVerifier
	certGossiper CertGossiper
	voteSigner   VoteSigner
	chainID      ids.ID
	nodeID       ids.NodeID

	// stakeSource (optional) makes finality STAKE-WEIGHTED instead of a raw voter
	// count (HIGH-3). When set (a value/PoS chain with unequal stake), a cert is
	// accepted only if its voters hold a strict ⅔ supermajority of stake at the
	// cert's height (VerifyWeighted), in addition to the α-of-K count. When nil,
	// finality is count-α and the chain MUST enforce equal stake at validator
	// admission (the documented invariant) — the node wires this for value chains.
	stakeSource StakeSource

	// setRootSource (optional) supplies the commitment to the active weighted
	// validator set at a block's height (the MEDIUM fix). When set, every
	// VotePosition this engine signs/assembles carries that set-root, so a cert
	// is cryptographically pinned to the exact set it was certified under and
	// cannot be re-verified against a different epoch's set. When nil, positions
	// carry ids.Empty (behavior identical to before set-root binding) — a chain
	// without epoch-versioned sets needs no binding.
	setRootSource ValidatorSetRootSource

	// strictPQ records that this chain runs under a STRICT post-quantum security
	// profile (set via WithStrictPQ, from config.Profile.IsStrict()). When true,
	// Mode() additionally requires a PQ cryptoWitness before reporting
	// ModeQuorumFinality, so the engine cannot report a quorum-finality
	// regime the chain cannot witness post-quantum.
	strictPQ bool

	// cryptoWitness (optional) is the post-quantum finality witness source a strict-PQ
	// chain wires (the SAME node-layer CryptoWitnessSource that upgrades an engine
	// QuorumCert into a quasar.WeightedQuorumCert — see quorum_cert_quasar.go). It is
	// REQUIRED for Mode() to report ModeQuorumFinality on a strict-PQ chain: without it the
	// cert path cannot produce the PQ (quasar) witness the profile demands, so the value-
	// DEX gate must not certify a quorum-finality regime that cannot be witnessed post-
	// quantum. On a non-strict chain it is unused. Injected like voteVerifier/stakeSource so
	// the gate reads the SAME field that delivers the witness — nil means "PQ witness not
	// plumbed", exactly the semantics ToQuasarCert already relies on.
	cryptoWitness CryptoWitnessSource

	// catchup (optional) is the engine's seam for runtime auto-recovery when it
	// falls behind — see Catchup. When a gossiped child or a verified cert
	// references a parent this node does not have, the engine asks catchup to
	// fetch the missing ancestors (requestCatchupLocked) instead of silently
	// dropping the child (the old behaviour that stranded a behind follower). nil
	// disables self-healing (legacy). The engine owns idempotency + rate-limiting
	// so the implementation can be a thin adapter onto the network getter.
	catchup Catchup

	// catchupRequested rate-limits ancestor fetches: it remembers the missing
	// parent IDs we have already asked for and when, so a re-gossip of the same
	// orphan (or many children of one missing parent) issues ONE fetch per
	// catchupCooldown, never a fetch storm. Keyed by the MISSING block ID.
	//
	// BOUNDED two ways (both fail-closed), because a Byzantine validator can stream
	// votes for forged random IDs that never arrive:
	//   - Reclaim-on-known: an entry is deleted the moment its block becomes TRACKED
	//     or DECIDED — at the accept span, the reject site, the sync reset, and at
	//     claimCatchupLocked's early returns (already-tracked / known-to-consensus).
	//     A block that actually arrives reclaims its slot; honest entries never pile
	//     up.
	//   - Hard cap + TTL: a forged ID that never arrives is never reclaimed above, so
	//     claimCatchupLocked refuses to grow the map past maxCatchupRequested — at
	//     the cap it sweeps entries older than catchupRequestTTL and, if still full
	//     (an active young flood), refuses the new claim. The map can never exceed
	//     maxCatchupRequested.
	catchupRequested map[ids.ID]time.Time

	// bufferedVotes parks signed accept/reject votes that arrived for a block this
	// node does not yet TRACK (the gossip race: a peer's vote can outrun the block
	// bytes). The old handleVote DROPPED such a vote — and because votes are only
	// solicited once, a dropped vote was lost forever, so a follower that missed
	// the block bytes could never reach the α-of-K quorum and the block wedged. We
	// instead BUFFER the vote (no signature work yet) and fetch the missing block
	// via the SAME catch-up seam used for a missing parent; when the block lands at
	// a tracking site, drainBufferedVotes replays these through the normal channel
	// path so each is signature-verified exactly as a live vote (buffering never
	// bypasses the gate). Keyed by the voted-on (missing) BlockID, and within each
	// block deduped by NodeID: at most ONE buffered vote per (BlockID, validator)
	// — a repeat from the same NodeID REPLACES its parked vote, never appends — so
	// the per-block slice is bounded by DISTINCT voters, not raw arrivals (the dual
	// of certVotes' NodeID keying; defeats single-Byzantine-ID buffer crowd-out).
	// Bounded by maxBufferedVoteBlocks distinct keys and maxBufferedVotesPerBlock
	// distinct NodeIDs per key (fail-closed: a new vote past a cap is dropped,
	// never evicting an existing one). Drained on track, deleted on decide — it
	// cannot leak.
	bufferedVotes map[ids.ID][]Vote

	// requestMissing is the engine's hook into the runtime's catch-up TRANSPORT
	// (Runtime.requestCatchup): "I am missing block `id` — fetch it from `from`".
	// It is the SAME one mechanism the missing-PARENT self-heal uses; the engine
	// (which does not hold the networkID) signals WHAT to fetch and the runtime
	// supplies the networkID + RequestAncestors round-trip. nil when no runtime is
	// wired (a bare engine in a unit test that drives delivery itself), in which
	// case a buffered vote simply waits for the block to be delivered by other
	// means. Set by NewRuntime; idempotency + rate-limiting stay in the engine
	// (claimCatchupLocked), so this is a thin signal, never a second transport.
	requestMissing func(missingID ids.ID, from ids.NodeID)

	// Logger for consensus events (nil-safe: uses log.Noop() if unset)
	log log.Logger
}

// CertGossiper broadcasts an assembled finality cert to validators. The node
// supplies the network implementation; the engine expresses WHAT (gossip this
// proof of α-of-K finality), the node decides HOW. Optional — a nil gossiper
// means the proposer finalizes locally on the α-of-K count without distributing
// the cert (followers then reach finality via their own collected votes once
// the topology gossips votes to all).
type CertGossiper interface {
	// GossipCert broadcasts the encoded finality cert for blockID to validators.
	GossipCert(chainID ids.ID, blockID ids.ID, certBytes []byte) error
}

// VoteSigner signs this node's accept vote over the canonical vote message so
// the signature can be collected into a QuorumCert. Backed by the node's
// validator key (the same key the VoteVerifier checks against). Optional: a
// single-validator engine does not gossip votes and needs no signer.
type VoteSigner interface {
	// SignVote returns this node's signature over message (the canonical vote
	// message for a position). The returned bytes are what a peer's
	// VoteVerifier will verify.
	SignVote(message []byte) ([]byte, error)
}

// Catchup is the engine's SOLE seam for "I am behind — fetch the block(s) I am
// missing". It is the one mechanism for runtime auto-recovery (decomplected from
// the finality path): the engine expresses WHAT (fetch the ancestors rooted at a
// missing block ID), the node decides HOW (a GetAncestors/Get round-trip on its
// existing network transport, delivering the fetched blocks back through
// HandleIncomingBlock).
//
// It is wired by the node (WithCatchup); when nil the engine simply does not
// self-heal a behind state (the legacy behaviour — a follower that falls behind
// at runtime is stranded). Idempotency and rate-limiting live in the ENGINE
// (requestCatchupLocked), so an implementation may be a thin, stateless adapter
// onto the network getter.
type Catchup interface {
	// RequestAncestors asks a peer to deliver the chain of blocks ending at
	// missingBlockID (the parent a gossiped child / verified cert referenced but
	// which this node does not have). `from` is the peer that advertised the
	// child — the natural source to fetch its parent from. chainID/networkID
	// scope the request to this chain's validator network. The fetched blocks are
	// expected to arrive via HandleIncomingBlock, at which point the formerly
	// orphaned child can be tracked, voted, and finalized.
	RequestAncestors(chainID ids.ID, networkID ids.ID, missingBlockID ids.ID, from ids.NodeID) error
}

// New creates an engine with default parameters.
// Apply options to configure dependencies.
//
// Example:
//
//	engine := New(WithVM(vm), WithProposer(proposer))
func New(opts ...Option) *Transitive {
	return NewWithConfig(DefaultConfig(), opts...)
}

// NewWithConfig creates an engine from explicit config plus options.
// Options are applied after config, allowing overrides.
//
// Example:
//
//	cfg := Config{Params: config.MainnetParams(), VM: vm}
//	engine := NewWithConfig(cfg, WithProposer(proposer))
func NewWithConfig(cfg Config, opts ...Option) *Transitive {
	// Scale buffers for burst mode — 1ms blocks produce 1000 blocks/sec,
	// so vote channels need depth to avoid back-pressure stalls.
	burst := cfg.Params.BlockTime <= time.Millisecond
	if cfg.VoteRequestBuffer == 0 {
		cfg.VoteRequestBuffer = 100
		if burst {
			cfg.VoteRequestBuffer = 4096
		}
	}
	if cfg.VoteBuffer == 0 {
		cfg.VoteBuffer = 1000
		if burst {
			cfg.VoteBuffer = 16384
		}
	}

	t := &Transitive{
		consensus:        NewChainConsensus(cfg.Params.K, cfg.Params.AlphaPreference, int(cfg.Params.Beta)),
		params:           cfg.Params,
		vm:               cfg.VM,
		proposer:         cfg.Proposer,
		bootstrapPhase:   true, // a fresh engine is initial-syncing until it reaches the frontier
		pendingBlocks:    make(map[ids.ID]*PendingBlock),
		finalizedByCert:  make(map[ids.ID]struct{}),
		certBytesByBlock: make(map[ids.ID][]byte),
		catchupRequested: make(map[ids.ID]time.Time),
		bufferedVotes:    make(map[ids.ID][]Vote),
		voteRequests:     make(chan VoteRequest, cfg.VoteRequestBuffer),
		votes:            make(chan Vote, cfg.VoteBuffer),
		pipelineSignal:   make(chan struct{}, 1),
	}

	for _, opt := range opts {
		opt(t)
	}

	if t.log == nil {
		t.log = log.Noop()
	}

	return t
}

// NewWithParams creates an engine with specific parameters.
func NewWithParams(params config.Parameters) *Transitive {
	cfg := DefaultConfig()
	cfg.Params = params
	return NewWithConfig(cfg)
}

// -----------------------------------------------------------------------------
// Lifecycle
// -----------------------------------------------------------------------------

// Start starts the engine.
func (t *Transitive) Start(ctx context.Context, _ bool) error {
	t.mu.Lock()
	defer t.mu.Unlock()

	if t.started {
		return ErrAlreadyStarted
	}

	// FAIL-CLOSED: a multi-validator engine (K>1) MUST have a vote verifier so
	// finality can be gated on a verifiable α-of-K quorum cert. Starting K>1
	// without one would leave no way to distinguish a real quorum from forged
	// votes — exactly the hole this change closes. A single-validator engine
	// (K==1) needs no verifier: its own accept is the quorum.
	if t.params.K > 1 && t.voteVerifier == nil {
		return ErrQuorumVerifierRequired
	}

	t.ctx, t.cancel = context.WithCancel(ctx)
	t.bootstrapped = true
	t.started = true

	// Capture ctx in local variable to avoid race with struct field access
	engineCtx := t.ctx

	t.wg.Add(4)
	go t.pollLoopWithCtx(engineCtx)
	go t.voteHandlerWithCtx(engineCtx)
	go t.pipelineLoop(engineCtx)
	go t.rePollLoopWithCtx(engineCtx)

	return nil
}

// StartWithID starts with a request ID.
func (t *Transitive) StartWithID(ctx context.Context, requestID uint32) error {
	return t.Start(ctx, requestID > 0)
}

// Stop stops the engine.
func (t *Transitive) Stop(ctx context.Context) error {
	// Cancel context first, outside the lock, to signal goroutines to exit.
	// This prevents deadlock where goroutines are blocked waiting for the lock
	// while we're holding the lock waiting for them to exit.
	t.mu.RLock()
	cancel := t.cancel
	t.mu.RUnlock()

	if cancel != nil {
		cancel()
	}

	// Wait for goroutines to exit before updating state.
	// This ensures clean shutdown without goroutine leaks.
	t.wg.Wait()

	t.mu.Lock()
	defer t.mu.Unlock()

	t.bootstrapped = false
	t.started = false
	return nil
}

// StopWithError stops with an error.
func (t *Transitive) StopWithError(ctx context.Context, _ error) error {
	return t.Stop(ctx)
}

// Context returns the engine's context.
func (t *Transitive) Context() context.Context {
	t.mu.RLock()
	defer t.mu.RUnlock()
	if t.ctx == nil {
		return context.Background()
	}
	return t.ctx
}

// IsBootstrapped returns bootstrap status.
func (t *Transitive) IsBootstrapped() bool {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.bootstrapped
}

// InBootstrapPhase reports whether the engine is still INITIAL-SYNCING (fetching
// + executing the chain from the network frontier). It is the gate the bootstrap
// accept authority (Runtime.AcceptBootstrapBlock) checks: a fetched-from-frontier
// block may finalize without an α-of-K cert ONLY while this is true. Once the node
// reaches the frontier (FinishBootstrap) it returns false and the bootstrap path is
// fail-closed — the live cert-gate is the only finalizer thereafter.
func (t *Transitive) InBootstrapPhase() bool {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.bootstrapPhase
}

// FinishBootstrap ends the bootstrap phase: the node has executed up to the
// discovered frontier and is entering live consensus. After this call the bootstrap
// accept path is fail-closed (InBootstrapPhase == false) and finality flows ONLY
// through the cert-witnessed α-of-K road. The node MUST call this exactly when it
// signals the chain bootstrapped (so the two transitions — "accept without a cert"
// and "no longer accept without a cert" — happen together). Idempotent.
func (t *Transitive) FinishBootstrap() {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.bootstrapPhase = false
}

// SyncState synchronizes consensus state with VM state.
// This is called by the syncer after RLP import or state sync to reconcile
// the consensus engine's lastAccepted pointer with the VM's actual state.
//
// This method:
//  1. Updates the consensus finalizedTip to match the VM's last accepted block
//  2. Clears any stale pending blocks that conflict with the new chain tip
//  3. Marks the engine as bootstrapped
//
// This is safe to call multiple times - it's idempotent.
func (t *Transitive) SyncState(ctx context.Context, lastAcceptedID ids.ID, height uint64) error {
	t.mu.Lock()
	defer t.mu.Unlock()

	// Update consensus state. A backward import is refused (ErrSyncStateRegression)
	// and must abort the whole reconcile — we do NOT clear pending blocks or flip
	// bootstrapped on a refused import, so a rejected regression is a clean no-op
	// rather than a partial state mutation.
	if t.consensus != nil {
		if err := t.consensus.SyncState(lastAcceptedID, height); err != nil {
			return err
		}
	}

	// Clear any pending blocks that are now stale (below the synced height)
	for blockID, pending := range t.pendingBlocks {
		if pending.ConsensusBlock != nil && pending.ConsensusBlock.height <= height {
			delete(t.pendingBlocks, blockID)
			// Votes parked for a stale (now-synced-past) block will never be drained
			// — drop them so a sync cannot leave buffered-vote residue.
			delete(t.bufferedVotes, blockID)
			// Same for its catch-up throttle entry: synced past ⇒ never re-fetched.
			delete(t.catchupRequested, blockID)
		}
	}

	// Ensure we're marked as bootstrapped
	t.bootstrapped = true

	return nil
}

// HealthCheck returns health stats.
func (t *Transitive) HealthCheck(ctx context.Context) (interface{}, error) {
	t.mu.RLock()
	defer t.mu.RUnlock()

	stats := t.consensus.Stats()
	stats["bootstrapped"] = t.bootstrapped
	stats["k"] = t.params.K
	stats["alpha"] = t.params.AlphaPreference
	stats["beta"] = t.params.Beta
	return stats, nil
}

// SetProposer sets the block proposer.
func (t *Transitive) SetProposer(proposer BlockProposer) {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.proposer = proposer
}

// SetEmitter sets the proposer (alias for SetProposer).
func (t *Transitive) SetEmitter(e BlockProposer) {
	t.SetProposer(e)
}

// SetVM sets the block builder.
func (t *Transitive) SetVM(vm BlockBuilder) {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.vm = vm
}

// -----------------------------------------------------------------------------
// Consensus operations
// -----------------------------------------------------------------------------

// AddBlock adds a block to consensus.
func (t *Transitive) AddBlock(ctx context.Context, blk *Block) error {
	return t.consensus.AddBlock(ctx, blk)
}

// CheckBlockProposal checks a block proposal for double-signing.
// Call this when receiving a block from a remote validator before adding it to consensus.
// Returns the evidence if the proposer equivocated, nil otherwise.
func (t *Transitive) CheckBlockProposal(proposerID ids.NodeID, height uint64, blockID ids.ID, blockData []byte) *slashing.Evidence {
	t.mu.RLock()
	detector := t.slashingDetector
	sdb := t.slashingDB
	t.mu.RUnlock()

	if detector == nil {
		return nil
	}

	// Reject proposals from jailed validators
	if sdb != nil && sdb.IsJailed(proposerID) {
		return &slashing.Evidence{
			Type:        slashing.DoubleSign,
			ValidatorID: proposerID,
			Height:      height,
		}
	}

	ev := detector.CheckBlock(proposerID, height, blockID, blockData)
	if ev != nil && sdb != nil {
		sdb.RecordEvidence(*ev)
	}
	return ev
}

// SlashingDB returns the slashing database, or nil if slashing is disabled.
func (t *Transitive) SlashingDB() *slashing.DB {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.slashingDB
}

// ProcessVote processes a vote.
func (t *Transitive) ProcessVote(ctx context.Context, blockID ids.ID, accept bool) error {
	return t.consensus.ProcessVote(ctx, blockID, accept)
}

// Poll conducts a poll.
func (t *Transitive) Poll(ctx context.Context, responses map[ids.ID]int) error {
	return t.consensus.Poll(ctx, responses)
}

// IsAccepted reports whether the block has been FINALIZED by this engine —
// committed through the SOLE cert-gated finalizer (AcceptWithCert), which is
// reachable only with a VerifiedQuorumCert. It is the engine's finality truth,
// NOT a vote count.
//
// CRITICAL: this no longer forwards consensus.IsAccepted. That is the raw α-of-K
// COUNT predicate — consensus.go sets block.accepted=true (and even populates its
// per-height finalized ledger) the instant acceptVotes>=alpha, with NO stake
// check. On a stake-weighted chain a low-stake/high-count coalition flips that
// count true WITHOUT a ⅔-stake supermajority, so reporting it would leak a
// finality claim the engine REFUSED to act on (no VM.Accept ran). Reading the
// engine's own finalizedByCert set — written only by AcceptWithCert — makes
// "accepted" mean exactly "finalized with a verified cert". A block stuck at
// count-α but lacking the cert is correctly NOT accepted here.
func (t *Transitive) IsAccepted(blockID ids.ID) bool {
	t.mu.RLock()
	defer t.mu.RUnlock()
	_, ok := t.finalizedByCert[blockID]
	return ok
}

// HasEnoughResponsesForRetry reports the LIVENESS signal: α-of-K validators have
// responded for the block (consensus.IsAccepted's raw count). This is the old
// IsAccepted count predicate, renamed to make its role unmistakable — it returns
// a RETRY signal, NEVER a finality verdict. A trigger may use it to decide
// whether it is worth calling TryAccept; it must NEVER itself finalize, and it
// does not increment blocksAccepted or touch the VM. Finality is decided solely
// by AcceptWithCert holding a VerifiedQuorumCert.
func (t *Transitive) HasEnoughResponsesForRetry(blockID ids.ID) bool {
	return t.consensus.IsAccepted(blockID)
}

// Preference returns preferred block.
func (t *Transitive) Preference() ids.ID {
	return t.consensus.Preference()
}

// GetBlock handles a block request.
func (t *Transitive) GetBlock(ctx context.Context, nodeID ids.NodeID, requestID uint32, blockID ids.ID) error {
	return nil
}

// Notify handles VM notifications.
func (t *Transitive) Notify(ctx context.Context, msg Message) error {
	t.mu.Lock()
	defer t.mu.Unlock()

	switch msg.Type {
	case PendingTxs:
		t.pendingBuildBlocks++
		return t.buildBlocksLocked(ctx)
	case StateSyncDone:
		return nil
	}
	return nil
}

// ReceiveVote queues a vote for processing.
// Returns false if the engine is not started (vote is dropped).
func (t *Transitive) ReceiveVote(vote Vote) bool {
	t.mu.RLock()
	started := t.started
	t.mu.RUnlock()

	if !started {
		return false
	}

	select {
	case t.votes <- vote:
		return true
	default:
		return false
	}
}

// Stats returns engine statistics.
func (t *Transitive) Stats() map[string]interface{} {
	t.mu.RLock()
	defer t.mu.RUnlock()

	stats := t.consensus.Stats()
	stats["blocks_built"] = t.blocksBuilt
	stats["blocks_accepted"] = t.blocksAccepted
	stats["blocks_rejected"] = t.blocksRejected
	stats["votes_sent"] = t.votesSent
	stats["votes_received"] = t.votesReceived
	stats["pending_blocks"] = len(t.pendingBlocks)
	stats["bootstrapped"] = t.bootstrapped
	return stats
}

// PendingBuildBlocks returns pending build count.
func (t *Transitive) PendingBuildBlocks() int {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.pendingBuildBlocks
}

// HasPendingBlock checks if a block is in the pending blocks map (built or received but not yet finalized).
// This is used by the Vote handler to determine if votes should be processed immediately
// (block exists) or buffered (block not yet available).
func (t *Transitive) HasPendingBlock(blockID ids.ID) bool {
	t.mu.RLock()
	defer t.mu.RUnlock()
	_, exists := t.pendingBlocks[blockID]
	return exists
}

// GetPendingBlock returns the VMBlock for a pending block if it exists.
// This allows the Vote handler to process votes for blocks that are in consensus
// but not yet verified/stored in the VM.
func (t *Transitive) GetPendingBlock(blockID ids.ID) (block.Block, bool) {
	t.mu.RLock()
	defer t.mu.RUnlock()
	pending, exists := t.pendingBlocks[blockID]
	if !exists || pending.VMBlock == nil {
		return nil, false
	}
	return pending.VMBlock, true
}

// -----------------------------------------------------------------------------
// Internal
// -----------------------------------------------------------------------------

func (t *Transitive) pollLoopWithCtx(ctx context.Context) {
	defer t.wg.Done()

	// Use BlockTime as poll interval — the engine must check finalization
	// at least as fast as blocks are produced. For 1ms blocks this means
	// 1ms polling; for mainnet 200ms blocks, 200ms polling.
	interval := t.params.BlockTime
	if interval <= 0 {
		interval = 50 * time.Millisecond // fallback
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if ctx.Err() != nil {
				return
			}
			t.processPendingBlocks()
		}
	}
}

// Re-poll backoff/cap constants. The re-poll loop is a RARE liveness BACKSTOP,
// not a hot loop: a block stuck because peers are BEHIND its parent is never
// recovered by re-soliciting the same block (those peers cannot vote), so we
// re-poll on a geometric schedule and then STOP, leaving recovery to the
// catch-up path (which fetches the missing chain so peers can vote).
const (
	// maxRePollBackoff caps the per-block re-poll interval. Starting from the base
	// RoundTO and doubling, the interval climbs RoundTO, 2·, 4·, … but never
	// exceeds this — so even a long-lived pending block is re-polled at most once
	// every maxRePollBackoff, turning a 250ms storm into a trickle.
	maxRePollBackoff = 16 * time.Second

	// maxRePollAttempts is the hard cap on re-poll attempts per block. After this
	// many re-solicitations the block is abandoned for re-poll (rePollAbandoned)
	// — it is never re-polled again (but stays pending, recoverable via cert or
	// catch-up). With doubling from a 250ms base, 8 attempts span
	// 0.25+0.5+1+2+4+8+16+16 ≈ 48s of backstop before giving up on re-poll.
	maxRePollAttempts = 8

	// catchupCooldown rate-limits ancestor fetches per missing parent: at most one
	// RequestAncestors per missing block ID per cooldown, so many children of one
	// missing parent (or repeated gossip of one orphan) cannot become a fetch
	// storm. The cooldown is generous because a fetch is a round-trip that, once it
	// lands, removes the orphan condition entirely.
	catchupCooldown = 2 * time.Second

	// maxCatchupRequested HARD-bounds the catchupRequested throttle map so a
	// Byzantine validator streaming votes for forged random BlockIDs (which never
	// arrive, so the delete-on-track/decide reclaim never fires for them) cannot
	// grow it without limit. Layer-1 (delete-on-track/decide + the early-return
	// reclaims in claimCatchupLocked) keeps honest entries from accumulating; this
	// hard cap is layer-2 (defence in depth) for the all-forged flood where layer-1
	// never reclaims. At the cap we first TTL-sweep, then fail closed (refuse the
	// new claim) rather than grow — so the map can never exceed this size.
	maxCatchupRequested = 4096

	// catchupRequestTTL is the age past which a catchupRequested entry is reclaimed
	// by the at-cap sweep. It is far beyond catchupCooldown: an honest fetch either
	// lands (and is reclaimed by delete-on-track) or is abandoned well inside 30s,
	// so a still-young entry at the cap signals an active forged flood — which we
	// answer by refusing new claims, never by unbounded growth.
	catchupRequestTTL = 30 * time.Second

	// maxBufferedVotesPerBlock caps how many votes we park for ONE missing block.
	// One vote per validator per block is the natural ceiling, so this must be ≥ the
	// largest supported validator set or the buffered fast-path silently drops
	// genuine votes 257..N and a net with α > cap cannot finalize from buffered votes
	// alone (recoverable only via re-poll). node/config/tokenomics.go defines a
	// 500-validator tier (and an unlimited tier); on a K=N / α=⌈⅔N⌉ chain with
	// N=500, α≈334. 512 covers the 500-tier with margin. A flood beyond the cap for
	// a single block ID is dropped (fail-closed) — the genuine α-of-K voters fit.
	// The bound stays small: 512 × maxBufferedVoteBlocks × ~64B ≈ 33MB worst case.
	maxBufferedVotesPerBlock = 512

	// maxBufferedVoteBlocks caps how many DISTINCT missing block IDs we will park
	// votes for at once. A spam stream of votes for never-delivered random block
	// IDs cannot grow the buffer past this many keys: once full, votes for a NEW
	// block ID are dropped (we do NOT evict an existing key — the simplest sound
	// bound). Happy-path keys are removed on drain (the block arrived) or on decide,
	// so this ceiling is only ever approached under adversarial junk.
	maxBufferedVoteBlocks = 1024
)

// rePollLoopWithCtx is the LIVENESS retry that prevents a terminal first-poll
// stall. The proposer issues exactly ONE RequestVotes when it builds a block
// (buildBlocksLocked) and runs finalizeOwnProposal ONCE right after; if at that
// instant the α-of-K signed votes have not yet arrived (the common case at
// genesis — peers are still bootstrapping, or the first PushQuery was dropped at
// the network boundary), the proposer's block sits in pendingBlocks with only its
// own self-vote and NOTHING re-solicits the missing votes. The finality poll loop
// (processPendingBlocks) only CHECKS consensus.IsAccepted; it never re-requests.
// So a single lagging validator at height 0 wedged finality forever — the devnet
// freeze. This loop implements the "poll-timeout re-request" the topology doc
// (quorum_topology.go) already promises but that was never wired.
//
// It is a pure liveness retry: it re-solicits votes and re-attempts cert
// assembly, and changes NOTHING about the finality predicate. A block still
// finalizes only on a verified α-of-K cert (multi-validator) or the 1-of-1 force
// (single-validator); a genuine minority still cannot and does not finalize.
//
// The ticker wakes on the base RoundTO, but each block is gated by its OWN
// EXPONENTIAL BACKOFF (PendingBlock.rePollBackoff, doubling per attempt, capped)
// and a HARD ATTEMPT CAP (maxRePollAttempts): a stuck block is re-driven on a
// geometric schedule and then ABANDONED for re-poll — turning the former
// fixed-cadence 250ms storm (the devnet self-DoS) into a bounded trickle. A
// behind follower is NOT recovered by re-polling (it cannot vote without the
// parent); it is recovered by the catch-up path that fetches the missing chain.
func (t *Transitive) rePollLoopWithCtx(ctx context.Context) {
	defer t.wg.Done()

	// Wake on the base round budget; per-block backoff decides whether a given
	// block is actually due. Fall back to a conservative 250ms if unset.
	interval := t.params.RoundTO
	if interval <= 0 {
		interval = 250 * time.Millisecond
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if ctx.Err() != nil {
				return
			}
			t.rePollAllPending(ctx, interval)
		}
	}
}

// rePollAllPending re-drives every undecided pending block that has gone a full
// re-poll interval without being re-solicited. For each such block it:
//
//  1. re-attempts finalization (tryFinalizeBlock) — idempotent; assembles +
//     gossips the cert if α signed votes are now present, so a follower that
//     missed the proposer's single cert-gossip still finalizes; and
//  2. if this node PROPOSED the block and a proposer transport is wired, re-issues
//     RequestVotes — re-sending the PushQuery so a laggard/peer that missed the
//     first poll receives the block + vote request again and can sign.
//
// Single-validator (K==1) engines never stall here (their own accept is the
// quorum, finalized synchronously), so the re-poll is a no-op for them.
//
// base is the base re-poll interval (RoundTO). Each block is gated by its OWN
// backoff window (rePollBackoff, doubling per attempt from base, capped at
// maxRePollBackoff) and abandoned once rePollAttempts reaches maxRePollAttempts.
func (t *Transitive) rePollAllPending(ctx context.Context, base time.Duration) {
	// K==1: no peer votes are ever needed; nothing to re-solicit.
	if t.consensus.K() <= 1 {
		return
	}

	now := time.Now()

	// Snapshot the blocks due for a re-poll under the lock, then act without it
	// (RequestVotes / tryFinalizeBlock take their own locks). A block is due once
	// it has been undecided for its CURRENT backoff window since the later of
	// ProposedAt and its last re-poll — so the FIRST re-poll waits one base
	// interval after proposal (giving the normal fast path time to finalize), then
	// the window DOUBLES each attempt, and after the cap the block is abandoned.
	type due struct {
		blockID   ids.ID
		blockData []byte
		ownProp   bool
	}
	var dueBlocks []due
	t.mu.Lock()
	for blockID, pending := range t.pendingBlocks {
		if pending.Decided || pending.rePollAbandoned {
			continue
		}
		// The window for THIS attempt: base for the first, doubling thereafter,
		// capped. rePollBackoff carries the PREVIOUS window (0 before the first).
		window := pending.rePollBackoff
		if window <= 0 {
			window = base
		}
		last := pending.lastRePoll
		if last.IsZero() {
			last = pending.ProposedAt
		}
		if now.Sub(last) < window {
			continue
		}

		// This block is due. Record the attempt, advance the backoff (double, cap),
		// and abandon once the cap is hit so it is NEVER re-polled again.
		pending.lastRePoll = now
		pending.rePollAttempts++
		next := window * 2
		if next > maxRePollBackoff {
			next = maxRePollBackoff
		}
		pending.rePollBackoff = next
		if pending.rePollAttempts >= maxRePollAttempts {
			pending.rePollAbandoned = true
			t.log.Warn("re-poll: block abandoned after attempt cap — not re-soliciting further (recoverable via cert/catch-up)",
				"blockID", blockID, "attempts", pending.rePollAttempts)
		}

		var data []byte
		if pending.VMBlock != nil {
			data = pending.VMBlock.Bytes()
		}
		dueBlocks = append(dueBlocks, due{blockID: blockID, blockData: data, ownProp: pending.IsOwnProposal})
	}
	proposer := t.proposer
	t.mu.Unlock()

	for _, d := range dueBlocks {
		// (1) Re-attempt finalization first: if α signed votes already arrived but
		// the single finalize attempt raced (or a follower missed the cert gossip),
		// this assembles + gossips the cert and commits now. Idempotent.
		t.tryFinalizeBlock(ctx, d.blockID)

		// (2) Proposer re-poll: re-send the vote request so a laggard re-receives
		// the block and votes. Only the proposer polls peers (followers learn the
		// block via gossip and broadcast their own votes); a follower short of
		// quorum recovers via the cert-gossip path that step (1) re-runs, or via
		// catch-up if it is behind the block's parent. The backoff above bounds how
		// often this fires; the cap stops it entirely for a terminally stuck block.
		if d.ownProp && proposer != nil {
			_ = proposer.RequestVotes(ctx, VoteRequest{
				BlockID:   d.blockID,
				BlockData: d.blockData,
			})
		}
	}
}

// claimCatchupLocked is the engine's idempotency + rate-limit GATE for "fetch
// this missing ancestor". It is the SINGLE decision point for whether a catch-up
// fetch should fire — the Runtime owns the actual network round-trip (it carries
// the networkID + transport). Caller holds t.mu.
//
// Returns true iff the caller should now issue exactly one RequestAncestors for
// missingID. It returns false (suppressing the fetch) when:
//   - no catchup transport is wired (legacy: a behind follower stays stranded),
//   - missingID is Empty (genesis/no parent — nothing to fetch),
//   - the block is already tracked or known to consensus (not actually missing),
//   - a fetch for this missing ID fired within catchupCooldown (throttle — so
//     many children of one missing parent, or repeated gossip of one orphan,
//     issue ONE fetch per cooldown, never a fetch storm).
//
// On a true return it records the throttle stamp, so the gate is self-arming:
// the caller does not need to remember anything. This is the catch-up analogue
// of the re-poll backoff — one mechanism, one place.
//
// The catchupRequested map is bounded TWO ways, both fail-closed:
//   - Layer 1 (reclaim-on-known): when missingID turns out to be already tracked
//     or known to consensus, its throttle entry can never be needed again, so we
//     delete it at those early returns (the dual of delete-on-track/decide at the
//     accept/reject/sync sites). An honest fetch that LANDS is reclaimed this way.
//   - Layer 2 (hard cap + TTL): a FORGED id that never arrives is never reclaimed
//     by layer 1, so before inserting a new entry at the cap we sweep entries
//     older than catchupRequestTTL; if the map is still at the cap (all young — an
//     active flood) we REFUSE the claim (no insert, no fetch). The map can never
//     exceed maxCatchupRequested.
func (t *Transitive) claimCatchupLocked(missingID ids.ID) bool {
	if t.catchup == nil || missingID == ids.Empty {
		return false
	}
	// Already have it tracked (or it just finalized) ⇒ not missing. A now-arrived
	// block can never be re-needed, so reclaim any throttle entry for it (layer 1).
	if _, tracked := t.pendingBlocks[missingID]; tracked {
		delete(t.catchupRequested, missingID)
		return false
	}
	if _, ok := t.consensus.GetBlock(missingID); ok {
		delete(t.catchupRequested, missingID)
		return false
	}
	now := time.Now()
	if last, ok := t.catchupRequested[missingID]; ok && now.Sub(last) < catchupCooldown {
		return false // throttled — one fetch per missing parent per cooldown
	}
	// HARD bound (layer 2): only when at the cap and inserting a NEW key. Sweep
	// entries past the TTL (an honest fetch resolves far inside catchupRequestTTL);
	// if still full afterward, every entry is young — an active forged-ID flood —
	// so fail closed: refuse the new claim rather than grow past the cap. The sweep
	// is O(size), bounded, and runs only at the cap (free in steady state).
	if _, existing := t.catchupRequested[missingID]; !existing && len(t.catchupRequested) >= maxCatchupRequested {
		for id, stamp := range t.catchupRequested {
			if now.Sub(stamp) >= catchupRequestTTL {
				delete(t.catchupRequested, id)
			}
		}
		if len(t.catchupRequested) >= maxCatchupRequested {
			return false // map saturated with active (young) entries — fail closed
		}
	}
	t.catchupRequested[missingID] = now
	return true
}

// pipelineLoop implements pipelined block production: as soon as a block is
// finalized, immediately build the next block. This decouples throughput from
// latency — with a 10-stage pipeline, a 10ms round produces 1 block/ms.
func (t *Transitive) pipelineLoop(ctx context.Context) {
	defer t.wg.Done()

	for {
		select {
		case <-ctx.Done():
			return
		case <-t.pipelineSignal:
			if ctx.Err() != nil {
				return
			}
			// A block was just finalized — immediately try to build the next one
			t.mu.Lock()
			if t.vm != nil {
				t.pendingBuildBlocks++
				_ = t.buildBlocksLocked(ctx)
			}
			t.mu.Unlock()
		}
	}
}

// signalPipeline wakes the pipeline goroutine to build the next block.
func (t *Transitive) signalPipeline() {
	select {
	case t.pipelineSignal <- struct{}{}:
	default: // already signaled
	}
}

func (t *Transitive) processPendingBlocks() {
	// Phase 1: Snapshot pending block IDs under t.mu (fast read lock).
	type candidate struct {
		blockID ids.ID
		vmBlock block.Block
	}
	t.mu.RLock()
	var candidates []candidate
	for blockID, pending := range t.pendingBlocks {
		if !pending.Decided {
			candidates = append(candidates, candidate{blockID: blockID, vmBlock: pending.VMBlock})
		}
	}
	t.mu.RUnlock()

	if len(candidates) == 0 {
		return
	}

	// Phase 2: classify each candidate WITHOUT holding t.mu (avoids nested lock).
	//
	// consensus.IsAccepted / IsRejected here are the LIVENESS TRIGGER, not the
	// finality authority: a true IsAccepted means "α-of-K voters responded — it is
	// worth ATTEMPTING to finalize", nothing more. The actual accept decision is
	// made by TryAccept in phase 3, which finalizes ONLY with a VerifiedQuorumCert
	// (the strict >⅔-of-stake gate). So a low-stake/high-count coalition that flips
	// IsAccepted here cannot finalize: TryAccept refuses it (ErrNoVerifiedQC) and
	// the block stays pending. Rejections carry no stake-safety concern (a block is
	// dropped, not finalized) and are committed inline.
	type blockAction struct {
		blockID ids.ID
		vmBlock block.Block
		accept  bool
	}
	var actions []blockAction
	for _, c := range candidates {
		if t.consensus.IsRejected(c.blockID) {
			actions = append(actions, blockAction{blockID: c.blockID, vmBlock: c.vmBlock, accept: false})
		} else if t.consensus.IsAccepted(c.blockID) {
			actions = append(actions, blockAction{blockID: c.blockID, vmBlock: c.vmBlock, accept: true})
		}
	}

	if len(actions) == 0 {
		return
	}

	// Phase 3: accepts go through the SOLE cert-gated path (TryAccept); rejects
	// are committed here. TryAccept is idempotent and takes its own lock — it
	// finalizes iff a verified cert exists, otherwise returns ErrNoVerifiedQC and
	// changes nothing (the block waits for the next tick). It also subsumes the
	// VM.Accept + SetPreference + pipeline-signal that the old phase 4 did inline,
	// so accepts no longer touch VM state from this loop at all.
	rejected := make([]blockAction, 0, len(actions))
	for _, action := range actions {
		if action.accept {
			_ = t.TryAccept(context.Background(), action.blockID)
			continue
		}
		rejected = append(rejected, action)
	}

	if len(rejected) == 0 {
		return
	}

	// Phase 4: commit the rejections (no cert required — a reject finalizes
	// nothing). Mirror the previous found/double-decide guard so a block already
	// decided by another trigger is not Reject'd twice.
	t.mu.Lock()
	ctx := t.ctx
	found := make([]bool, len(rejected))
	for i, action := range rejected {
		pending, exists := t.pendingBlocks[action.blockID]
		if !exists || pending.Decided {
			continue
		}
		found[i] = true
		pending.Decided = true
		t.blocksRejected++
		delete(t.pendingBlocks, action.blockID)
		// Drop any votes parked for a now-rejected block (it will never be tracked
		// to drain them) so the buffer cannot leak.
		delete(t.bufferedVotes, action.blockID)
		// A rejected (decided) block is no longer "missing" — reclaim its catch-up
		// throttle entry so catchupRequested stays bounded.
		delete(t.catchupRequested, action.blockID)
	}
	t.mu.Unlock()

	for i, action := range rejected {
		if !found[i] {
			continue // already decided by another trigger
		}
		if action.vmBlock != nil {
			action.vmBlock.Reject(ctx)
		}
	}
}

func (t *Transitive) voteHandlerWithCtx(ctx context.Context) {
	defer t.wg.Done()

	for {
		select {
		case <-ctx.Done():
			return
		case vote := <-t.votes:
			// Check context again before processing to avoid work after shutdown
			if ctx.Err() != nil {
				return
			}
			t.handleVote(vote)
		}
	}
}

func (t *Transitive) handleVote(vote Vote) {
	// Collect state under t.mu, release before calling consensus methods
	// to avoid nested lock (t.mu -> c.mu) deadlock.
	t.mu.Lock()
	t.votesReceived++
	pending, exists := t.pendingBlocks[vote.BlockID]
	detector := t.slashingDetector
	sdb := t.slashingDB
	ctx := t.ctx

	if !exists {
		// A vote for an ALREADY-FINALIZED block needs nothing: the block is decided
		// (removed from pendingBlocks, recorded in finalizedByCert). Do NOT buffer or
		// fetch it — that would re-park a late/duplicate vote AFTER acceptWithCertCore
		// cleared the buffer, leaking a slot for a block that will never re-track to
		// drain it. Drop it (the finality cert already exists).
		if _, finalized := t.finalizedByCert[vote.BlockID]; finalized {
			t.mu.Unlock()
			return
		}
		// VOTE FOR A BLOCK WE DO NOT YET TRACK — the gossip race that wedged the
		// write path: a peer's vote outran the block bytes. The old code DROPPED it
		// here, and since votes are solicited only once, the drop was permanent —
		// the missing-bytes follower could never reach α-of-K and the block never
		// Accepted. Instead BUFFER the vote (bounded, no signature work yet) and ask
		// the catch-up seam to FETCH the missing block, exactly as the missing-parent
		// path does. When the block lands at a tracking site, drainBufferedVotes
		// replays each parked vote through the normal channel path, where it is
		// signature-verified like any live vote (buffering NEVER bypasses the gate;
		// a forged/unsigned parked vote costs one map slot and is dropped on replay).
		accepted := t.bufferVoteLocked(vote)
		// GATE the fetch on buffer acceptance. If the buffer REFUSED the vote (a cap
		// was hit), do NOT fetch — a fetch for a vote we did not park is pure
		// amplification: there is nothing buffered for the fetched block to drain
		// into. Fetching ONLY for parked votes gives a bounded aggregate fetch rate:
		// at most min(maxBufferedVoteBlocks, maxCatchupRequested) distinct fetches in
		// flight (one per parked-and-claimed missing ID), each re-fireable at most
		// once per catchupCooldown. That is the global fetch ceiling — it falls out
		// of bounding BOTH bufferedVotes and catchupRequested.
		var fetch func(missingID ids.ID, from ids.NodeID)
		if accepted {
			fetch = t.requestMissing
		}
		t.mu.Unlock()
		// Fire the fetch WITHOUT the lock: requestMissing routes to
		// Runtime.requestCatchup, which claims its OWN lock for the idempotency gate
		// (claimCatchupLocked) and then does the RequestAncestors round-trip. Calling
		// it while still holding t.mu would re-enter the lock (non-reentrant) and
		// deadlock. nil ⇒ no runtime wired (bare engine) OR the buffer rejected the
		// vote (no payoff in fetching); either way the vote waits / is dropped.
		if fetch != nil {
			fetch(vote.BlockID, vote.NodeID)
		}
		return
	}

	var height uint64
	if pending.ConsensusBlock != nil {
		height = pending.ConsensusBlock.height
	}

	// Equivocation + jail checks before counting the vote.
	// Detector and DB use their own locks; safe to call under t.mu
	// since there is no lock ordering conflict (t.mu -> detector.mu is one direction only).
	if detector != nil {
		if ev := detector.CheckVote(vote.NodeID, height, vote.BlockID, vote.Accept); ev != nil {
			if sdb != nil {
				sdb.RecordEvidence(*ev)
			}
			t.mu.Unlock()
			return
		}
		if sdb != nil && sdb.IsJailed(vote.NodeID) {
			t.mu.Unlock()
			return
		}
	}

	// AUTHENTICATE THE VOTE (multi-validator). A vote is counted toward the
	// quorum ONLY if its signature verifies over the block's position AND the
	// vote's own decision (accept→accept-message, reject→reject-message). This:
	//   - makes consensus.acceptVotes count only REAL validator accepts, so
	//     IsAccepted() is truthful (it cannot be inflated by forged/unsigned
	//     votes) and stays in lock-step with the assemblable cert;
	//   - blocks forged-accept finality (an outsider key, or a real validator's
	//     signature lifted from a DIFFERENT position/decision, fails here);
	//   - blocks forged-reject censorship (an unauthenticated reject is dropped,
	//     strictly safer than the prior unauthenticated reject path).
	//
	// The former `effectiveAccept = vote.Accept || IsOwnProposal` REJECT→ACCEPT
	// flip is DELETED: vote.Accept is authoritative once the signature checks
	// out. Single-validator engines (no verifier) skip authentication — the sole
	// validator's self-vote is the quorum and carries no signature.
	if t.voteVerifier != nil {
		pos := t.blockPositionLocked(pending, vote.BlockID)
		msg := canonicalVoteMessageFor(pos, vote.Accept)
		// Resolve the voter's pubkey at the block's P-CHAIN epoch height (RESIDUAL-B),
		// the same height the position's set-root commits to.
		if len(vote.Signature) == 0 || !t.voteVerifier.VerifyVote(vote.NodeID, msg, vote.Signature, t.epochHeightLocked(pending)) {
			// Unsigned or invalid: not a real vote from this validator at this
			// position/decision. Drop it — count nothing.
			t.mu.Unlock()
			return
		}
	}

	accept := vote.Accept
	var voteCount int
	if accept {
		pending.VoteCount++
		voteCount = pending.VoteCount
		// Record the signed accept vote toward this block's quorum cert so the
		// engine can assemble + gossip the α-of-K witness once the threshold is
		// reached. (Reject votes are not certifiable — a finality cert proves
		// acceptance — they only drive the rejection path.)
		t.recordCertVoteLocked(pending, vote)
	} else {
		pending.RejectCount++
	}
	t.mu.Unlock()

	if err := t.consensus.ProcessVote(ctx, vote.BlockID, accept); err != nil {
		return
	}
	_ = t.consensus.Poll(ctx, map[ids.ID]int{vote.BlockID: voteCount})

	// Finalize: if consensus reached the α-of-K accept quorum, assemble the
	// cert, gossip it, and call VM.Accept().
	t.tryFinalizeBlock(ctx, vote.BlockID)
}

// bufferVoteLocked parks a vote for a not-yet-tracked block, enforcing both
// bounds (fail-closed) AND one-vote-per-(block, validator) dedup. Caller holds
// t.mu. It does NO signature work — a parked vote is verified only when
// drainBufferedVotes replays it through handleVote.
//
// Dedup invariant: at most ONE buffered vote per (BlockID, NodeID). If a vote for
// this {BlockID, NodeID} pair is already parked, it is REPLACED in place rather
// than appended, so the per-block slice is bounded by DISTINCT NodeIDs, not raw
// arrival count. This is the dual of the certVotes NodeID-keyed dedup that
// recordCertVoteLocked applies to live votes — same keying, mirrored onto the
// slice. It closes the single-Byzantine-ID crowd-out: one NodeID can occupy at
// most ONE slot per block, so it cannot flood maxBufferedVotesPerBlock junk
// entries and crowd genuine validators' votes out of the buffer fast-path.
//
// Returns accepted=true iff a vote for this block is now parked/represented; the
// caller uses this to GATE the catch-up fetch. A same-(block, node) REPLACEMENT
// still returns true: there is a parked vote for that block to drain into, so the
// fetch is still warranted. accepted=false only when a cap dropped the vote — and
// a vote we refused to even buffer must NOT trigger a fetch (firing one for a
// dropped vote is pure amplification with no payoff — nothing is parked for the
// fetched block to drain into).
//   - Per-block cap: if maxBufferedVotesPerBlock DISTINCT NodeIDs are already
//     parked for this block ID, a vote from a NEW NodeID is dropped (the real
//     α-of-K voters fit well within; a replacement of an existing NodeID never
//     hits the cap — it does not grow the slice).
//   - Total-keys cap: if this is a NEW block ID and maxBufferedVoteBlocks distinct
//     IDs are already parked, the new key is dropped (we never evict an existing
//     key — the simplest sound bound; existing keys drain on track or delete on
//     decide).
func (t *Transitive) bufferVoteLocked(vote Vote) (accepted bool) {
	existing, seen := t.bufferedVotes[vote.BlockID]
	if !seen && len(t.bufferedVotes) >= maxBufferedVoteBlocks {
		return false // total distinct-block ceiling reached — fail closed
	}
	// Dedup by NodeID (dual of certVotes): if this validator already has a vote
	// parked for this block, replace it in place — never append a second. This is
	// what bounds the slice by distinct NodeIDs and defeats single-ID crowd-out.
	for i := range existing {
		if existing[i].NodeID == vote.NodeID {
			existing[i] = vote
			return true // a vote for this block is parked — fetch still warranted
		}
	}
	if len(existing) >= maxBufferedVotesPerBlock {
		return false // per-block ceiling reached — fail closed
	}
	t.bufferedVotes[vote.BlockID] = append(existing, vote)
	return true
}

// drainBufferedVotes replays every vote parked for blockID now that the block is
// tracked. It removes the slice under t.mu, deletes the key (so the buffer cannot
// leak for a block that did arrive), and re-enqueues each vote via ReceiveVote —
// the SAME channel path a live vote takes. Re-enqueueing (rather than calling
// handleVote inline) avoids re-entrant locking and keeps ONE code path: each
// replayed vote re-enters handleVote with the block now tracked, so it is
// signature-verified exactly as a live vote (a forged/unsigned parked vote is
// dropped at the gate; it never counts). Called at every block-tracking site. If
// the vote channel is full, ReceiveVote returns false and the vote is dropped —
// acceptable: the periodic re-poll / re-gossip will re-deliver it.
func (t *Transitive) drainBufferedVotes(blockID ids.ID) {
	t.mu.Lock()
	parked := t.bufferedVotes[blockID]
	delete(t.bufferedVotes, blockID)
	t.mu.Unlock()

	for _, vote := range parked {
		t.ReceiveVote(vote)
	}
}

// recordCertVoteLocked records a distinct SIGNED accept vote toward this
// block's quorum cert. Caller holds t.mu. A vote with no signature is ignored
// for cert purposes (it still counts toward the plain accept tally in
// handleVote) — only signed votes can witness a cert. Verification of the
// signature happens at assembly time (assembleCertLocked) so a single bad
// signature cannot poison the map; de-dup is by NodeID.
func (t *Transitive) recordCertVoteLocked(pending *PendingBlock, vote Vote) {
	if len(vote.Signature) == 0 {
		return
	}
	if pending.certVotes == nil {
		pending.certVotes = make(map[ids.NodeID]SignedVote)
	}
	pending.certVotes[vote.NodeID] = SignedVote{
		NodeID:    vote.NodeID,
		Accept:    true,
		Signature: append([]byte(nil), vote.Signature...),
	}
}

// recordOwnVoteLocked signs THIS node's accept vote for blockID and records it
// into the block's cert set. Caller holds t.mu. No-op when no voteSigner is
// configured (single-validator / K==1 finality needs no cert). The proposer
// (and any node casting its own accept locally) is one of the α signers, so its
// signature belongs in the cert.
func (t *Transitive) recordOwnVoteLocked(pending *PendingBlock, blockID ids.ID) {
	if t.voteSigner == nil {
		return
	}
	pos := t.blockPositionLocked(pending, blockID)
	sig, err := t.voteSigner.SignVote(CanonicalVoteMessage(pos))
	if err != nil {
		t.log.Warn("failed to sign own accept vote for cert", "blockID", blockID, "error", err)
		return
	}
	t.recordCertVoteLocked(pending, Vote{
		BlockID:   blockID,
		NodeID:    t.nodeID,
		Accept:    true,
		Signature: sig,
		ParentID:  pos.ParentID,
		Round:     pos.Round,
	})
}

// pChainHeighter is the subset of block.SignedBlock the engine needs to pin a
// block's validator-set epoch: the P-CHAIN height the block was proposed at. A
// proposervm block satisfies it; a bare VM block does not (epoch height 0 →
// fail-closed on the K>1 finality path). Defined locally so the engine depends
// only on the one method it reads, not the whole SignedBlock surface.
type pChainHeighter interface {
	PChainHeight() uint64
}

// pChainHeightOf extracts the P-chain height a VM block was proposed at, or 0 if
// the block does not carry one (pre-fork / no proposervm wrapper). This is the
// SOLE place the engine reads PChainHeight off a VM block, so every consensus
// Block records the same epoch the proposervm signed.
func pChainHeightOf(b block.Block) uint64 {
	if ph, ok := b.(pChainHeighter); ok {
		return ph.PChainHeight()
	}
	return 0
}

// PChainHeightOfForTest exposes the engine's block→P-chain-height boundary read to
// a test in ANOTHER module (the node's chains package), so it can prove the
// node-layer wrapper delivers the REAL epoch height (not 0) through the EXACT
// function the engine uses. Exported only for that cross-module test reach.
func PChainHeightOfForTest(b block.Block) uint64 { return pChainHeightOf(b) }

// epochHeightLocked returns the P-CHAIN height the block's weighted validator set
// is pinned to — the SINGLE height used for the set-root commitment, the
// ⅔-by-stake tally, AND per-voter pubkey resolution (membership, pubkey,
// set-root, stake ALL read set@H, MEDIUM-1/CRITICAL-1/RESIDUAL-B). It is the
// block's recorded P-chain height, NOT its value-chain height. Caller holds t.mu.
func (t *Transitive) epochHeightLocked(pending *PendingBlock) uint64 {
	if pending != nil && pending.ConsensusBlock != nil {
		return pending.ConsensusBlock.pChainHeight
	}
	return 0
}

// blockPositionLocked returns the consensus position a block's votes/cert bind
// to. Caller holds t.mu.
func (t *Transitive) blockPositionLocked(pending *PendingBlock, blockID ids.ID) VotePosition {
	var parentID ids.ID
	var height uint64
	if pending.ConsensusBlock != nil {
		parentID = pending.ConsensusBlock.parentID
		height = pending.ConsensusBlock.height
	}
	// Stamp the active weighted-validator-set commitment at the block's P-CHAIN
	// EPOCH height (the MEDIUM-1 / CRITICAL-1 fix) — NOT its value-chain height.
	// Every path that builds a position — sign (recordOwnVoteLocked), assemble +
	// verify (assembleCertLocked), incoming-vote/cert verify — routes through
	// here, so they all bind the SAME root for a given block: a cert is pinned to
	// the exact set it was certified under. The epoch height is the proposervm's
	// PChainHeight, the only height that is (i) ≤ the current P-chain height (so
	// platformvm.GetValidatorSet does NOT errUnfinalizedHeight) and (ii) embedded
	// in the signed block so every honest node derives the IDENTICAL set/root.
	// nil source ⟹ Empty root (the fixed-set no-op).
	var setRoot ids.ID
	if t.setRootSource != nil {
		setRoot = t.setRootSource.ValidatorSetRoot(t.epochHeightLocked(pending))
	}
	return VotePosition{
		ChainID:          t.chainID,
		Height:           height,
		Round:            pending.Round,
		BlockID:          blockID,
		ParentID:         parentID,
		ValidatorSetRoot: setRoot,
	}
}

// TrackOwnProposalForTest inserts blk as a verified own-proposal pending block —
// the SAME state buildBlocksLocked establishes for a locally built block — and
// returns the canonical VotePosition followers must sign. It exists so a test in
// ANOTHER module (the node's chains package) can drive a REAL VM block (e.g. the
// node's P-chain-height-stamping wrapper block) through the engine's actual
// vote→assemble→verify→finalize path. It is exported ONLY for that cross-module
// test reach; it is not part of the consensus runtime surface.
//
// It is NOT a finality shortcut: it records the proposer's own signed accept
// (recordOwnVoteLocked) and a single self-vote toward the count exactly as
// production does, and it NEVER calls ForceAccept. A block tracked here finalizes
// (K>1) only when enough real signed peer votes arrive to assemble a cert that
// VERIFIES under the wired verifier (and clears the ⅔-stake predicate when a
// stake source is wired) — the genuine BFT path.
//
// The load-bearing line is `pChainHeightOf(blk)`: it captures the block's P-CHAIN
// epoch height off the VM block through the SAME boundary the production
// buildBlocksLocked uses, so a test can prove the boundary delivers the real
// height (not 0) end to end. The returned position's set-root is stamped at that
// epoch height (blockPositionLocked), so a follower signs — and the verifier
// resolves pubkeys at — the LIVE set@H.
func (t *Transitive) TrackOwnProposalForTest(ctx context.Context, blk block.Block, round uint32) VotePosition {
	cb := &Block{
		id:           blk.ID(),
		parentID:     blk.ParentID(),
		height:       blk.Height(),
		timestamp:    blk.Timestamp().Unix(),
		data:         blk.Bytes(),
		pChainHeight: pChainHeightOf(blk), // the boundary capture under test (b2)
	}
	_ = t.consensus.AddBlock(ctx, cb)
	_ = t.consensus.ProcessVote(ctx, blk.ID(), true)
	t.mu.Lock()
	pb := &PendingBlock{
		ConsensusBlock: cb,
		VMBlock:        blk,
		ProposedAt:     time.Now(),
		VoteCount:      1,
		Round:          round,
		Decided:        false,
		IsOwnProposal:  true,
	}
	t.pendingBlocks[blk.ID()] = pb
	t.recordOwnVoteLocked(pb, blk.ID())
	pos := t.blockPositionLocked(pb, blk.ID())
	t.mu.Unlock()
	// Replay any votes a peer parked for this block before we tracked it (a
	// follower could have seen a peer's vote for our own block before our build
	// signal). Drain after unlock — drainBufferedVotes takes t.mu.
	t.drainBufferedVotes(blk.ID())
	return pos
}

// assembleCertLocked attempts to assemble a verified QuorumCert from the signed
// accept votes collected for blockID. Caller holds t.mu. Returns the cert (and
// caches it on pending) iff:
//   - a vote verifier is configured (multi-validator finality), AND
//   - at least alpha distinct votes verify under it.
//
// Each collected vote's signature is verified here; votes that fail are dropped
// from the candidate set so one forged vote cannot block a real quorum, and the
// cert is only built from VERIFIED votes — Assemble + the subsequent Verify
// then re-check distinctness and the threshold. Returns nil if the verified
// quorum is not yet present (the proposer keeps waiting / re-requesting — this
// is the liveness path, NOT a force).
func (t *Transitive) assembleCertLocked(pending *PendingBlock, blockID ids.ID) *QuorumCert {
	if pending.cert != nil {
		return pending.cert
	}
	if t.voteVerifier == nil {
		return nil
	}
	alpha := t.consensus.Alpha()
	if alpha <= 0 {
		return nil
	}
	pos := t.blockPositionLocked(pending, blockID)
	message := CanonicalVoteMessage(pos)
	// The epoch height pins every per-voter pubkey resolution + the stake tally to
	// the SAME P-chain height the position's set-root commits to (MEDIUM-1).
	epochHeight := t.epochHeightLocked(pending)

	verified := make([]SignedVote, 0, len(pending.certVotes))
	for _, sv := range pending.certVotes {
		if t.voteVerifier.VerifyVote(sv.NodeID, message, sv.Signature, epochHeight) {
			verified = append(verified, sv)
		}
	}
	if uint32(len(verified)) < uint32(alpha) {
		return nil
	}
	cert, err := AssembleQuorumCert(pos, uint32(alpha), verified)
	if err != nil {
		return nil
	}
	// Defence in depth: the cert we just built must verify under our own
	// verifier before we treat it as a finality witness (catches any assembly
	// invariant drift). Assemble already enforced distinctness + threshold.
	// On a stake-weighted chain it must ALSO clear the ⅔-of-stake supermajority
	// (HIGH-3) — the count quorum alone is not finality when stake is unequal, so
	// we keep WAITING (return nil) until enough STAKE has voted, never forcing.
	if t.stakeSource != nil {
		if err := cert.VerifyWeighted(t.voteVerifier, t.stakeSource, epochHeight); err != nil {
			return nil
		}
	} else if err := cert.Verify(t.voteVerifier, epochHeight); err != nil {
		return nil
	}
	pending.cert = cert
	return cert
}

// TryAccept is the ONE entry every finality trigger calls — a vote arrived, a
// re-poll fired, the pending queue changed, a block was built/verified, a poll
// timeout ticked. It is the single funnel onto the cert-gated acceptance path:
//
//	cert, err := <obtain a VerifiedQuorumCert for blockID>
//	if err != nil { return err }   // ErrNoVerifiedQC: not final yet, retry later
//	return t.AcceptWithCert(ctx, blockID, cert)
//
// A raw α-of-K COUNT is NOT an acceptance authority here. It is a LIVENESS
// signal: it may BRING us to TryAccept (the poll loop / vote handler call this
// when consensus signals "enough responses"), but TryAccept finalizes ONLY if a
// VerifiedQuorumCert can be produced — i.e. the votes assemble into a cert that
// clears VerifyWeighted's strict >⅔-of-stake gate. If not, TryAccept returns
// ErrNoVerifiedQC and changes nothing: the block stays pending+undecided and the
// trigger retries on its next tick. No count, no callback, no "enough voters
// responded" can finalize without the cert. This is the structural HIGH-3 fix.
//
//   - K>1: the cert is assembled+verified from the collected SIGNED votes
//     (assembleVerifiedCertLocked → BuildVerifiedQuorumCert semantics). The
//     verified cert is gossiped so followers finalize on the same proof.
//   - K==1: there are no peers; the sole validator's own accept IS the 1-of-1
//     quorum. We commit it through the per-height guard (ForceAccept, which
//     refuses K>1) and wrap the resulting decision as a 1-of-1 VerifiedQuorumCert
//     so even this path finalizes ONLY through AcceptWithCert — one finalizer.
func (t *Transitive) TryAccept(ctx context.Context, blockID ids.ID) error {
	t.mu.Lock()
	pending, exists := t.pendingBlocks[blockID]
	if !exists || pending.Decided {
		t.mu.Unlock()
		return nil // nothing to accept (gone or already finalized) — not an error
	}
	singleValidator := t.consensus.K() == 1

	if singleValidator {
		// 1-of-1 quorum. Commit through the per-height guard (ForceAccept refuses
		// K>1) so a K==1 node still cannot finalize two blocks at one height, then
		// authorize the shared finalizer with a 1-of-1 verified cert.
		if err := t.consensus.ForceAccept(blockID); err != nil {
			t.mu.Unlock()
			t.log.Crit("ForceAccept refused — engine misconfigured (K>1 reached single-validator path)",
				"blockID", blockID, "error", err)
			return err
		}
		cert := t.buildSingleValidatorCertLocked(pending, blockID)
		t.mu.Unlock()
		return t.AcceptWithCert(ctx, blockID, cert)
	}

	// K>1: a verified α-of-K cert is the ONLY authority. Build+verify it from the
	// collected signed votes. nil ⇒ the verified ⅔-stake quorum is not present
	// yet ⇒ ErrNoVerifiedQC ⇒ liveness retry (NOT a finalize).
	cert, ok := t.assembleVerifiedCertLocked(pending, blockID)
	if !ok {
		t.mu.Unlock()
		return ErrNoVerifiedQC
	}
	var certBytes []byte
	if b, err := cert.Cert().MarshalBinary(); err == nil {
		certBytes = b
	}
	chainID := t.chainID
	gossiper := t.certGossiper
	t.mu.Unlock()

	// Distribute the finality proof so followers finalize on the same verifiable
	// witness (not a fast-follow guess). Best effort — local finality already
	// holds via the verified cert about to be committed.
	if gossiper != nil && certBytes != nil {
		_ = gossiper.GossipCert(chainID, blockID, certBytes)
	}

	return t.AcceptWithCert(ctx, blockID, cert)
}

// tryFinalizeBlock is a thin compatibility shim onto TryAccept for the
// peer-quorum triggers (poll-due, vote-handler). It exists so those call sites
// read as "try to finalize"; all real logic — and the cert gate — is in
// TryAccept. A consensus COUNT reaching α is what brings us here, but TryAccept
// finalizes only with a VerifiedQuorumCert; ErrNoVerifiedQC is the normal
// "wait" answer and is intentionally swallowed (the trigger retries next tick).
func (t *Transitive) tryFinalizeBlock(ctx context.Context, blockID ids.ID) {
	_ = t.TryAccept(ctx, blockID)
}

// finalizeOwnProposal is the proposer-side trigger after building its own block.
//
// THE FREEZE THIS USED TO "FIX" — AND HOW IT IS NOW FIXED WITHOUT SELF-FINALITY:
// the old version FORCE-ACCEPTED the proposer's own block on its lone self-vote
// (self-finality — a value could finalize with NO α-of-K agreement, so an
// equivocating proposer could fork the chain). DELETED for K>1. The freeze is
// now solved STRUCTURALLY by the vote-distribution topology (integration.go):
// followers gossip their SIGNED accept votes to ALL validators, the proposer
// assembles + gossips the cert, and finality comes via the verified cert.
//
// This is now just another trigger: it routes to TryAccept like every other
// trigger. K==1 finalizes via the 1-of-1 cert; K>1 finalizes IFF a verified
// α-of-K cert exists. Never forces a K>1 block on the lone self-vote.
func (t *Transitive) finalizeOwnProposal(ctx context.Context, blockID ids.ID) {
	// The own block is now tracked (buildBlocksLocked inserted it before calling
	// here, with the lock released). Replay any votes a peer parked for it before
	// our build signal so they count toward this attempt. Lock-free (drain takes
	// t.mu); every caller invokes this without holding t.mu.
	t.drainBufferedVotes(blockID)
	t.mu.RLock()
	pending, exists := t.pendingBlocks[blockID]
	own := exists && pending.IsOwnProposal && !pending.Decided
	t.mu.RUnlock()
	if !own {
		return
	}
	_ = t.TryAccept(ctx, blockID)
}

// assembleVerifiedCertLocked builds the FINALITY AUTHORITY TOKEN for blockID from
// the collected signed accept votes, or reports that no verified quorum exists
// yet. Caller holds t.mu. It delegates the predicate to assembleCertLocked
// (assemble + signature-verify + VerifyWeighted's strict >⅔-of-stake gate — the
// SINGLE place the stake predicate lives) and wraps the verified result. ok=false
// (zero cert) ⇒ the verified ⅔-stake quorum is not present yet ⇒ the caller must
// NOT finalize (it returns ErrNoVerifiedQC and the trigger retries). There is no
// other in-engine producer of the token for the multi-validator path, so the
// count road has no way to manufacture finality.
func (t *Transitive) assembleVerifiedCertLocked(pending *PendingBlock, blockID ids.ID) (VerifiedQuorumCert, bool) {
	cert := t.assembleCertLocked(pending, blockID)
	if cert == nil {
		return VerifiedQuorumCert{}, false
	}
	// assembleCertLocked has already run VerifyWeighted/Verify before caching the
	// cert, so promotion is safe; wrapVerifiedCert refuses only a nil cert.
	return wrapVerifiedCert(cert)
}

// buildSingleValidatorCertLocked produces the 1-of-1 VerifiedQuorumCert for the
// K==1 path so the single-validator node finalizes through the SAME sole
// finalizer (AcceptWithCert) as every other path — one finalization road. Caller
// holds t.mu and has already committed the decision via ForceAccept (which itself
// passes the per-height guard). On a K==1 chain α==1 and the sole validator's own
// signed accept (recordOwnVoteLocked, captured at build time) is the entire
// quorum; assembleCertLocked verifies that single signature and the (trivially
// satisfied) stake gate. If a verifier/signer is not wired (a pure single-node
// dev chain with no vote crypto), there is no signature to certify — we authorize
// the commit with a degenerate non-zero token whose cert carries the position,
// because ForceAccept already established the 1-of-1 finality and the per-height
// guard already prevented equivocation. This degenerate token exists ONLY for
// K==1 and can never arise for K>1 (TryAccept's K>1 branch never calls here).
func (t *Transitive) buildSingleValidatorCertLocked(pending *PendingBlock, blockID ids.ID) VerifiedQuorumCert {
	// Prefer the VERIFIED 1-of-1 cert: when vote crypto is wired the proposer
	// recorded its own signed accept (recordOwnVoteLocked), so assembleCertLocked
	// verifies that single signature (and the trivially-met stake gate) and we
	// finalize on a real witness — even on a single-validator chain.
	if cert, ok := t.assembleVerifiedCertLocked(pending, blockID); ok {
		return cert
	}
	// Only when NO vote crypto is wired (a pure single-node dev chain, voteVerifier
	// nil) is there no signature to certify. Synthesize the 1-of-1 finality witness
	// from the position: ForceAccept has ALREADY committed this block through the
	// per-height guard (the real single-node safety gate — one block per height, no
	// branching), so the token only authorizes the VM.Accept. This branch is
	// K==1-only (both callers gate on K()==1) and verifier-nil-only, so it can
	// never substitute for a real α-of-K cert on any multi-validator chain.
	if t.voteVerifier != nil {
		// Verifier wired but the self-vote did not assemble (should not happen on a
		// healthy K==1 node). Do NOT fabricate a witness when crypto is available —
		// return zero so AcceptWithCert refuses and the next trigger retries.
		return VerifiedQuorumCert{}
	}
	pos := t.blockPositionLocked(pending, blockID)
	return VerifiedQuorumCert{qc: &QuorumCert{
		Version:   QuorumCertVersion,
		Type:      QCFinality,
		Position:  pos,
		Threshold: 1,
	}}
}

// AcceptWithCert is the SOLE function that can finalize a block. It is impossible
// to call without a VerifiedQuorumCert value, and a zero VerifiedQuorumCert
// (cert==nil) is refused — so the ONLY way to reach VM.Accept is to first hold a
// cert that cleared the finality predicate (BuildVerifiedQuorumCert /
// assembleVerifiedCertLocked / the verified incoming-cert path). The old
// finalizePendingLocked body lives here unchanged; the difference is that it can
// no longer be reached by any count-only road — the type system enforces it.
//
// Idempotent: subsequent calls find pending.Decided=true and no-op.
//
// It signals the pipeline to build the next block on success — the right thing
// for an OUT-OF-BAND finalize (a vote/cert arrived async, or the poll loop fired)
// where nothing else is driving production. The synchronous in-build-loop path
// (buildBlocksLocked) instead uses acceptWithCertCore(..., signalNext=false): it
// is already inside the build loop, so re-signaling would spawn a SECOND
// concurrent builder and race the VM's block counter (the K=1 burst regression).
func (t *Transitive) AcceptWithCert(ctx context.Context, blockID ids.ID, cert VerifiedQuorumCert) error {
	return t.acceptWithCertCore(ctx, blockID, cert, true)
}

// acceptWithCertCore is the one finalization body. signalNext controls only
// whether it wakes the pipeline afterward (see AcceptWithCert). Everything that
// makes finality safe — the zero-cert refusal, the Decided/idempotency guard,
// the VM Accept+SetPreference ordering — is identical on both call paths.
func (t *Transitive) acceptWithCertCore(ctx context.Context, blockID ids.ID, cert VerifiedQuorumCert, signalNext bool) error {
	if cert.IsZero() {
		// No verified witness ⇒ no finality. This is the structural guarantee:
		// even an internal caller cannot finalize by passing a zero cert.
		return ErrNoVerifiedQC
	}

	t.mu.Lock()
	pending, exists := t.pendingBlocks[blockID]
	if !exists || pending.Decided {
		t.mu.Unlock()
		return nil
	}
	pending.Decided = true
	t.blocksAccepted++
	delete(t.pendingBlocks, blockID)
	// A decided block can never need parked votes again — drop any so the buffer
	// cannot leak for an already-finalized block (the drain handles the happy path
	// where the block arrived; this handles the already-decided path).
	delete(t.bufferedVotes, blockID)
	// Likewise reclaim its catch-up throttle entry: a finalized block is no longer
	// "missing", so no future fetch for it can fire — keep catchupRequested bounded.
	delete(t.catchupRequested, blockID)
	// Record engine-level finality. This is the ONLY place finalizedByCert is
	// written, and it is written ONLY here — the sole finalizer, reachable only
	// with a non-zero VerifiedQuorumCert. Transitive.IsAccepted reads this set, so
	// engine finality is exactly "AcceptWithCert ran for this block", never the
	// count-driven consensus.IsAccepted.
	t.finalizedByCert[blockID] = struct{}{}
	// Retain the cert that just authorized this finalize so this node can serve it
	// to a peer catching up (CertForBlock). cert is the SAME VerifiedQuorumCert that
	// cleared the predicate above; marshaling it here — the one finalizer — captures
	// every finalize path (local assembly, incoming gossiped cert, K==1) in one spot.
	if qc := cert.Cert(); qc != nil {
		if b, err := qc.MarshalBinary(); err == nil {
			t.storeServedCertLocked(blockID, b)
		}
	}
	t.mu.Unlock()

	if pending.VMBlock != nil {
		_ = pending.VMBlock.Accept(ctx)
	}

	// SetPreference must follow Accept to keep VM and consensus in sync.
	t.mu.RLock()
	vm := t.vm
	t.mu.RUnlock()
	if vm != nil {
		if err := vm.SetPreference(ctx, blockID); err != nil {
			t.log.Crit("SetPreference failed after Accept — forcing preference",
				"blockID", blockID,
				"error", err)
			t.consensus.ForcePreference(blockID)
		}
	}

	// Pipeline: block finalized → immediately build next (out-of-band callers
	// only; the in-build-loop caller continues its own loop and passes false).
	if signalNext {
		t.signalPipeline()
	}

	// MED-6: bound the slashing detector's per-height vote/block maps to a
	// sliding window below the finalized height. Equivocation is only
	// actionable near the tip; retaining every height's records unboundedly is a
	// memory-exhaustion vector. Prune everything older than the window.
	t.pruneSlashingBelowWindow()
	return nil
}

// slashingRetentionHeights is how many heights below the finalized tip the
// slashing detector retains vote/block records for. Equivocation evidence is
// only useful near the tip (a fork is attempted at or above the last finalized
// height); older records cannot prove a NEW double-vote and are pruned to bound
// memory. 1024 heights is ample for cross-validator timing skew at any block
// time while keeping the maps O(window·validators).
const slashingRetentionHeights = uint64(1024)

// pruneSlashingBelowWindow drops slashing records older than the retention
// window below the finalized height. No-op when no detector is wired or when the
// chain has not advanced past the window.
func (t *Transitive) pruneSlashingBelowWindow() {
	t.mu.RLock()
	detector := t.slashingDetector
	t.mu.RUnlock()
	if detector == nil {
		return
	}
	fh, set := t.consensus.GetFinalizedHeight()
	if !set || fh <= slashingRetentionHeights {
		return
	}
	detector.PruneBelow(fh - slashingRetentionHeights)
}

// DrainAccepted attempts to finalize any pending block consensus has SIGNALLED
// as accepted. Called from the ForwardVMNotifications loop after each Notify.
//
// consensus.IsAccepted is the LIVENESS trigger only (α-of-K responded — worth
// attempting). The finality decision is made by TryAccept, which finalizes ONLY
// with a VerifiedQuorumCert (the >⅔-stake gate) and otherwise returns
// ErrNoVerifiedQC and changes nothing. This closes the previously count-ONLY
// finalize road here: a block drained from this loop now finalizes through the
// exact same cert-gated path (AcceptWithCert) as every other trigger — no count
// can VM.Accept without the cert.
func (t *Transitive) DrainAccepted(ctx context.Context) {
	t.mu.RLock()
	candidates := make([]ids.ID, 0, len(t.pendingBlocks))
	for id, pending := range t.pendingBlocks {
		if !pending.Decided && t.consensus.IsAccepted(id) {
			candidates = append(candidates, id)
		}
	}
	t.mu.RUnlock()

	for _, id := range candidates {
		_ = t.TryAccept(ctx, id)
	}
}

func (t *Transitive) buildBlocksLocked(ctx context.Context) error {
	if t.vm == nil {
		return nil
	}

	for t.pendingBuildBlocks > 0 {
		vmBlock, err := t.vm.BuildBlock(ctx)
		if err != nil {
			t.log.Error("BuildBlock failed, will retry next tick",
				"error", err,
				"pendingBuildBlocks", t.pendingBuildBlocks)
			// Do NOT decrement pendingBuildBlocks — the request is still
			// outstanding and will be retried on the next Notify or pipeline tick.
			return nil
		}
		t.pendingBuildBlocks--

		t.blocksBuilt++

		// Skip if already tracked
		if _, exists := t.pendingBlocks[vmBlock.ID()]; exists {
			continue
		}

		consensusBlock := &Block{
			id:           vmBlock.ID(),
			parentID:     vmBlock.ParentID(),
			height:       vmBlock.Height(),
			timestamp:    vmBlock.Timestamp().Unix(),
			data:         vmBlock.Bytes(),
			pChainHeight: pChainHeightOf(vmBlock), // epoch for the weighted set (MEDIUM-1)
		}

		// Verify BEFORE consensus — prevents accepting invalid blocks in K=1 mode
		// where self-vote causes immediate acceptance. If Verify fails, the block
		// is never added to consensus, so IsAccepted cannot return true for it.
		t.mu.Unlock()
		if err := vmBlock.Verify(ctx); err != nil {
			t.mu.Lock()
			continue
		}

		// Now add to consensus and self-vote.
		addErr := t.consensus.AddBlock(ctx, consensusBlock)
		if addErr == nil {
			_ = t.consensus.ProcessVote(ctx, vmBlock.ID(), true)
			_ = t.consensus.Poll(ctx, map[ids.ID]int{vmBlock.ID(): 1})
		}
		t.mu.Lock()

		if addErr != nil {
			continue
		}

		// Track the block as a pending own proposal — ALWAYS, including the K=1
		// single-node case. We no longer VM.Accept inline here: finalization for
		// every block (K=1 or K>1) goes through the SOLE cert-gated finalizer via
		// TryAccept below, so there is exactly one acceptance road. In K=1 TryAccept
		// commits the 1-of-1 quorum (ForceAccept) and finalizes through
		// AcceptWithCert; the call is synchronous (see below) so it runs before the
		// next BuildBlock.
		pb := &PendingBlock{
			ConsensusBlock: consensusBlock,
			VMBlock:        vmBlock,
			ProposedAt:     time.Now(),
			VoteCount:      1,
			Decided:        false,
			IsOwnProposal:  true,
		}
		t.pendingBlocks[vmBlock.ID()] = pb
		// The proposer is one of the α signers: record its OWN signed accept
		// vote into the cert set so the assembled cert includes it (its
		// ProcessVote above counted it toward acceptVotes; this puts the
		// matching SIGNED record in certVotes so count and cert agree).
		t.recordOwnVoteLocked(pb, vmBlock.ID())

		// Gossip the block + request peer votes. These calls are done while
		// holding t.mu — keep them short (msg creation + queue, no waiting).
		proposerWired := t.proposer != nil
		if proposerWired {
			proposal := BlockProposal{
				BlockID:   vmBlock.ID(),
				BlockData: vmBlock.Bytes(),
				Height:    vmBlock.Height(),
				ParentID:  vmBlock.ParentID(),
			}
			t.proposer.Propose(ctx, proposal)
			voteReq := VoteRequest{
				BlockID:    vmBlock.ID(),
				BlockData:  vmBlock.Bytes(),
				Validators: nil,
			}
			t.proposer.RequestVotes(ctx, voteReq)
		}

		// Self-finalize the just-built own block through the SOLE cert-gated
		// finalizer (AcceptWithCert). The block was verified locally above, so the
		// proposer has committed to its correctness; waiting only on peer Chits to
		// drive finality causes the lux-devnet stall when Chits arrive late.
		blockID := vmBlock.ID()
		singleValidator := t.consensus.K() == 1

		if singleValidator {
			// K==1: the sole validator's accept IS the 1-of-1 quorum. Commit it
			// through the per-height guard (ForceAccept) and build the 1-of-1
			// VerifiedQuorumCert — both UNDER the lock we already hold — then
			// release the lock only for the VM Accept/SetPreference inside
			// AcceptWithCert. Doing this INLINE (not via the async poll loop) is
			// load-bearing for correctness, not just speed: the VM advances its
			// parent only on SetPreference, and the per-height guard requires each
			// block's parent to equal the finalized tip and its height to be the
			// strict successor. If the next BuildBlock ran before this block's
			// SetPreference, it would read a stale parent and the commit would be
			// refused (ErrParentNotFinalizedTip), stalling the chain. Finalizing
			// here keeps build→accept→SetPreference→next-build in lockstep, exactly
			// as the pre-collapse inline accept did — only now routed through the
			// one finalizer.
			if err := t.consensus.ForceAccept(blockID); err != nil {
				t.log.Crit("ForceAccept refused — engine misconfigured (K>1 reached single-validator path)",
					"blockID", blockID, "error", err)
			} else {
				cert := t.buildSingleValidatorCertLocked(pb, blockID)
				t.mu.Unlock()
				// signalNext=false: we are inside the build loop, which drives the
				// next build itself. Re-signaling would spawn a concurrent builder
				// and gap the VM block counter (the K=1 burst-throughput stall).
				_ = t.acceptWithCertCore(ctx, blockID, cert, false)
				t.mu.Lock()
			}
		} else if proposerWired {
			// K>1 with a network proposer: attempt finality now via the verified
			// α-of-K cert. The cert may already be assemblable from collected votes;
			// if not, TryAccept no-ops (ErrNoVerifiedQC) and the poll loop / cert
			// gossip retry. NEVER forces a K>1 block on the lone self-vote.
			t.mu.Unlock()
			t.finalizeOwnProposal(ctx, blockID)
			t.mu.Lock()
		}
	}
	return nil
}

// -----------------------------------------------------------------------------
// Transport (network layer interface)
// -----------------------------------------------------------------------------

// Transport handles message transport.
type Transport[ID comparable] interface {
	Send(ctx context.Context, to string, msg interface{}) error
	Receive(ctx context.Context) (interface{}, error)
}