record delivered-but-unsettled requests as delivered_pending_settlement

devshard/cmd/devshardctl/escrow_rotator.go

@@ -109,6 +109,7 @@ func (g *Gateway) rotateEscrowsOnce() {
 	}
 	if !pocActive {
 		g.finishBridgeEscrows(snapshot, settings)
+		g.retrySettlements(context.Background(), snapshot.EpochIndex, settings)
 	}
 }
 
@@ -151,6 +152,7 @@ func (g *Gateway) prepareBridgeEscrows(snapshot ChainPhaseSnapshot, settings Gat
 			if err != nil {
 				log.Printf("escrow_rotation_regular_retire_failed epoch=%d model=%q escrow=%s error=%v", epoch, model.ModelID, devshard.ID, err)
 				settleFailed++
+				g.registerSettlementRetry(devshard.ID, epoch+1)
 			} else if settledOnChain {
 				settled++
 			}
@@ -219,6 +221,7 @@ func (g *Gateway) finishBridgeEscrows(snapshot ChainPhaseSnapshot, settings Gate
 			if err != nil {
 				log.Printf("escrow_rotation_temp_retire_failed epoch=%d model=%q escrow=%s error=%v", epoch, model.ModelID, devshard.ID, err)
 				settleFailed++
+				g.registerSettlementRetry(devshard.ID, epoch+1)
 			} else if settledOnChain {
 				settled++
 			}
@@ -411,19 +414,16 @@ func (g *Gateway) settleDevshardOnChain(ctx context.Context, id string, req admi
 		return nil, err
 	}
 	log.Printf("devshard_settle_key_loaded escrow=%s settler=%s key_env=%q", id, signer.Address(), privateKeyEnv)
-	if rt.proxy.sm.Phase() != types.PhaseSettlement {
-		g.finalizeMu.Lock()
-		log.Printf("gateway_finalize_lock_acquired escrow=%s path=rotation_settle", id)
-		if err := rt.session.Finalize(ctx); err != nil {
-			g.finalizeMu.Unlock()
-			log.Printf("devshard_settle_failed escrow=%s stage=finalize error=%q", id, err.Error())
-			return nil, err
-		}
-		g.finalizeMu.Unlock()
-		log.Printf("devshard_settle_finalize_completed escrow=%s phase=%s", id, sessionPhaseLabel(rt.proxy.sm.Phase()))
-	} else {
-		log.Printf("devshard_settle_finalize_skipped escrow=%s phase=%s", id, sessionPhaseLabel(rt.proxy.sm.Phase()))
-	}
+	g.finalizeMu.Lock()
+	log.Printf("gateway_finalize_lock_acquired escrow=%s path=rotation_settle", id)
+	// Always finalize: the PhaseSettlement branch re-collects a short quorum, so a retry after a validator returns completes instead of re-broadcasting the same short proof.
+	finalizeErr := rt.session.Finalize(ctx)
+	g.finalizeMu.Unlock()
+	if finalizeErr != nil {
+		log.Printf("devshard_settle_failed escrow=%s stage=finalize error=%q", id, finalizeErr.Error())
+		return nil, finalizeErr
+	}
+	log.Printf("devshard_settle_finalize_completed escrow=%s phase=%s", id, sessionPhaseLabel(rt.proxy.sm.Phase()))
 	settlement, err := rt.proxy.settlementJSON()
 	if err != nil {
 		log.Printf("devshard_settle_failed escrow=%s stage=settlement_json error=%q", id, err.Error())
@@ -445,5 +445,59 @@ func (g *Gateway) settleDevshardOnChain(ctx context.Context, id string, req admi
 		return nil, err
 	}
 	log.Printf("devshard_settle_submitted escrow=%s tx_hash=%s settler=%s", id, result.TxHash, result.Settler)
+	g.clearSettlementRetry(id)
 	return result, nil
 }
+
+// registerSettlementRetry queues a devshard whose settlement failed for retry through deadlineEpoch (inclusive), keeping the earliest deadline if already queued.
+func (g *Gateway) registerSettlementRetry(id string, deadlineEpoch uint64) {
+	g.settlementRetryMu.Lock()
+	defer g.settlementRetryMu.Unlock()
+	if g.settlementRetry == nil {
+		g.settlementRetry = make(map[string]uint64)
+	}
+	if _, exists := g.settlementRetry[id]; !exists {
+		g.settlementRetry[id] = deadlineEpoch
+	}
+}
+
+func (g *Gateway) clearSettlementRetry(id string) {
+	g.settlementRetryMu.Lock()
+	defer g.settlementRetryMu.Unlock()
+	delete(g.settlementRetry, id)
+}
+
+func (g *Gateway) pendingSettlementRetries() map[string]uint64 {
+	g.settlementRetryMu.Lock()
+	defer g.settlementRetryMu.Unlock()
+	if len(g.settlementRetry) == 0 {
+		return nil
+	}
+	out := make(map[string]uint64, len(g.settlementRetry))
+	for id, deadline := range g.settlementRetry {
+		out[id] = deadline
+	}
+	return out
+}
+
+// retrySettlements re-attempts failed settlements until success or the deadline epoch; past that the on-chain reaper takes over. Success clears the entry inside settleDevshardOnChain.
+func (g *Gateway) retrySettlements(ctx context.Context, currentEpoch uint64, settings GatewaySettings) {
+	if !settings.EscrowRotation.SettlementEnabled {
+		return
+	}
+	for id, deadlineEpoch := range g.pendingSettlementRetries() {
+		if currentEpoch > deadlineEpoch {
+			g.clearSettlementRetry(id)
+			log.Printf("settlement_retry_expired escrow=%s epoch=%d deadline=%d", id, currentEpoch, deadlineEpoch)
+			continue
+		}
+		attemptCtx, cancel := context.WithTimeout(ctx, autoSettlementAttemptTimeout)
+		_, err := gatewaySettleDevshardOnChain(g, attemptCtx, id, adminSettleEscrowRequest{})
+		cancel()
+		if err != nil {
+			log.Printf("settlement_retry_failed escrow=%s epoch=%d deadline=%d error=%v", id, currentEpoch, deadlineEpoch, err)
+			continue
+		}
+		log.Printf("settlement_retry_succeeded escrow=%s epoch=%d", id, currentEpoch)
+	}
+}

devshard/cmd/devshardctl/escrow_settlement_retry_test.go

@@ -0,0 +1,91 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func settlementEnabledSettings() GatewaySettings {
+	return GatewaySettings{EscrowRotation: EscrowRotationSettings{SettlementEnabled: true}}
+}
+
+func stubSettleDevshardOnChain(t *testing.T, fn func(id string) (*SettleDevshardEscrowResult, error)) {
+	t.Helper()
+	saved := gatewaySettleDevshardOnChain
+	t.Cleanup(func() { gatewaySettleDevshardOnChain = saved })
+	gatewaySettleDevshardOnChain = func(g *Gateway, _ context.Context, id string, _ adminSettleEscrowRequest) (*SettleDevshardEscrowResult, error) {
+		result, err := fn(id)
+		if err == nil {
+			g.clearSettlementRetry(id) // mirror the real settleDevshardOnChain, which clears on success
+		}
+		return result, err
+	}
+}
+
+func TestSettlementRetry_RetriesUntilSuccessThenClears(t *testing.T) {
+	attempts := 0
+	stubSettleDevshardOnChain(t, func(string) (*SettleDevshardEscrowResult, error) {
+		attempts++
+		if attempts < 3 {
+			return nil, errors.New("insufficient signatures")
+		}
+		return &SettleDevshardEscrowResult{}, nil
+	})
+
+	gateway := &Gateway{}
+	gateway.registerSettlementRetry("escrow-1", 5)
+	settings := settlementEnabledSettings()
+
+	gateway.retrySettlements(context.Background(), 5, settings)
+	require.Contains(t, gateway.pendingSettlementRetries(), "escrow-1", "still failing, kept for retry")
+
+	gateway.retrySettlements(context.Background(), 5, settings)
+	require.Contains(t, gateway.pendingSettlementRetries(), "escrow-1")
+
+	gateway.retrySettlements(context.Background(), 5, settings)
+	require.NotContains(t, gateway.pendingSettlementRetries(), "escrow-1", "cleared once settled")
+	require.Equal(t, 3, attempts)
+}
+
+func TestSettlementRetry_ExpiresPastDeadlineWithoutRetrying(t *testing.T) {
+	attempts := 0
+	stubSettleDevshardOnChain(t, func(string) (*SettleDevshardEscrowResult, error) {
+		attempts++
+		return nil, errors.New("insufficient signatures")
+	})
+
+	gateway := &Gateway{}
+	gateway.registerSettlementRetry("escrow-1", 5)
+
+	gateway.retrySettlements(context.Background(), 6, settlementEnabledSettings())
+
+	require.NotContains(t, gateway.pendingSettlementRetries(), "escrow-1", "dropped past deadline")
+	require.Equal(t, 0, attempts, "no on-chain call past the deadline")
+}
+
+func TestSettlementRetry_NoopWhenSettlementDisabled(t *testing.T) {
+	attempts := 0
+	stubSettleDevshardOnChain(t, func(string) (*SettleDevshardEscrowResult, error) {
+		attempts++
+		return &SettleDevshardEscrowResult{}, nil
+	})
+
+	gateway := &Gateway{}
+	gateway.registerSettlementRetry("escrow-1", 5)
+
+	gateway.retrySettlements(context.Background(), 5, GatewaySettings{})
+
+	require.Contains(t, gateway.pendingSettlementRetries(), "escrow-1", "left untouched while settlement disabled")
+	require.Equal(t, 0, attempts)
+}
+
+func TestRegisterSettlementRetry_KeepsEarliestDeadline(t *testing.T) {
+	gateway := &Gateway{}
+	gateway.registerSettlementRetry("escrow-1", 5)
+	gateway.registerSettlementRetry("escrow-1", 9)
+
+	require.Equal(t, uint64(5), gateway.pendingSettlementRetries()["escrow-1"])
+}

devshard/cmd/devshardctl/gateway.go

@@ -62,6 +62,8 @@ type Gateway struct {
 	settlementInFlight    map[string]struct{}
 	replenishmentMu       sync.Mutex
 	replenishmentInFlight map[string]struct{}
+	settlementRetryMu     sync.Mutex
+	settlementRetry       map[string]uint64 // devshard id -> last epoch to retry on-chain settlement (inclusive)
 	mu                    sync.Mutex
 	roundRobinSeed        atomic.Uint64
 }

devshard/cmd/devshardctl/redundancy.go

@@ -3304,12 +3304,14 @@ func (e *Redundancy) finishRaceOutcome(ctx context.Context, attempts []*inflight
 				e.recordGatewayTimeoutAction(inf, params, timeoutResultKind(result, inf), "completed", "none")
 			}
 		}
+		// Only relabels accounting/metrics; escrow and the returned error stay untouched.
+		terminal := e.resolveTerminalDecision(attempts, winnerNonce, failed)
 		if hostErr := hostApplicationErrorFromAttempts(attempts, winnerNonce); hostErr != nil {
 			captureAllAttemptsFailedRequest(ctx, e.devshardID, params, hostErr)
-			logRequestStage(ctx, "request_failed", "escrow", e.devshardID, "error", hostErr)
-			e.recordGatewayRequestOutcome(params.Model, "failed", gatewayRequestFailureReason(failed))
-			e.completeAccountingRequest(ctx, 0, decision, "failed")
-			e.logRequestSettled(ctx, 0, decision, "failed")
+			logRequestStage(ctx, terminal.stage, "escrow", e.devshardID, "error", hostErr)
+			e.recordGatewayRequestOutcome(params.Model, terminal.outcome, terminal.reason)
+			e.completeAccountingRequest(ctx, terminal.nonce, decision, terminal.outcome)
+			e.logRequestSettled(ctx, terminal.nonce, decision, terminal.outcome)
 			e.checkEscrowMissing(ctx, attempts)
 			return hostErr
 		}
@@ -3321,10 +3323,10 @@ func (e *Redundancy) finishRaceOutcome(ctx context.Context, attempts []*inflight
 			errMsg = (&nonStreamingReducedMaxTokensTimeoutError{}).Error()
 		}
 		captureAllAttemptsFailedRequest(ctx, e.devshardID, params, fmt.Errorf("%s", errMsg))
-		logRequestStage(ctx, "request_failed", "escrow", e.devshardID, "error", errMsg)
-		e.recordGatewayRequestOutcome(params.Model, "failed", gatewayRequestFailureReason(failed))
-		e.completeAccountingRequest(ctx, 0, decision, "failed")
-		e.logRequestSettled(ctx, 0, decision, "failed")
+		logRequestStage(ctx, terminal.stage, "escrow", e.devshardID, "error", errMsg)
+		e.recordGatewayRequestOutcome(params.Model, terminal.outcome, terminal.reason)
+		e.completeAccountingRequest(ctx, terminal.nonce, decision, terminal.outcome)
+		e.logRequestSettled(ctx, terminal.nonce, decision, terminal.outcome)
 		e.checkEscrowMissing(ctx, attempts)
 		if opts.nonStreamingReducedTokenTimeout {
 			return &nonStreamingReducedMaxTokensTimeoutError{}
@@ -3463,6 +3465,46 @@ func (e *Redundancy) resolvedWinnerNonce(attempts []*inflight, winnerNonce uint6
 	return 0
 }
 
+// winnerDeliveredPendingSettlement reports whether the crowned winner streamed clean content to the client but its nonce never finished — the executor delivered a response yet produced no applied finish, so it cannot settle. Not a failure from the client's view.
+func (e *Redundancy) winnerDeliveredPendingSettlement(attempts []*inflight, winnerNonce uint64) bool {
+	if winnerNonce == 0 {
+		return false
+	}
+	winner := inflightByNonce(attempts, winnerNonce)
+	if winner == nil || winner.probe {
+		return false
+	}
+	if e.session.IsNonceFinished(winner.nonce) || errors.Is(winner.processErr, types.ErrStateHashMismatch) {
+		return false
+	}
+	return winner.err == nil && winner.contentChunks.Load() > 0 && !isFailedStreamAttempt(winner)
+}
+
+type terminalDecision struct {
+	nonce   uint64
+	outcome string
+	stage   string
+	reason  string
+}
+
+// resolveTerminalDecision picks the accounting outcome for a request that failed to reach a finished nonce: delivered_pending_settlement when the winner delivered content, otherwise failed.
+func (e *Redundancy) resolveTerminalDecision(attempts []*inflight, winnerNonce uint64, failed []*inflight) terminalDecision {
+	if e.winnerDeliveredPendingSettlement(attempts, winnerNonce) {
+		return terminalDecision{
+			nonce:   winnerNonce,
+			outcome: "delivered_pending_settlement",
+			stage:   "request_delivered_pending_settlement",
+			reason:  "none",
+		}
+	}
+	return terminalDecision{
+		nonce:   0,
+		outcome: "failed",
+		stage:   "request_failed",
+		reason:  gatewayRequestFailureReason(failed),
+	}
+}
+
 func fallbackSuspiciousWinner(attempts []*inflight) *inflight {
 	for _, inf := range attempts {
 		if inf == nil || inf.probe || !inf.suspicious {