feat(billing): 完成异步结算与请求执行闭环

统一任务成功、Attempt 与结算 Outbox 的事务边界，增加多实例安全结算、人工复核、请求幂等与执行租约。钱包决策使用九位精确金额，并通过审计保护约束保留流水事实；同时补充管理接口、指标与 PostgreSQL 集成测试。

apps/api/internal/runner/billing_mode_test.go

@@ -0,0 +1,27 @@
+package runner
+
+import "testing"
+
+func TestNormalizedBillingEngineMode(t *testing.T) {
+	t.Parallel()
+	if got := normalizedBillingEngineMode(""); got != "observe" {
+		t.Fatalf("empty mode = %q", got)
+	}
+	if got := normalizedBillingEngineMode("ENFORCE"); got != "enforce" {
+		t.Fatalf("enforce mode = %q", got)
+	}
+	if got := normalizedBillingEngineMode("hold"); got != "hold" {
+		t.Fatalf("hold mode = %q", got)
+	}
+}
+
+func TestBillingItemsFixedTotalKeepsNineDecimalPlaces(t *testing.T) {
+	t.Parallel()
+	items := []any{
+		map[string]any{"amount": "0.000000001"},
+		map[string]any{"amount": float64(0.000000002)},
+	}
+	if got := billingItemsFixedTotal(items).String(); got != "0.000000003" {
+		t.Fatalf("total = %s", got)
+	}
+}

apps/api/internal/runner/billing_settlement_worker.go

@@ -0,0 +1,96 @@
+package runner
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
+	"github.com/google/uuid"
+)
+
+const billingSettlementPollInterval = time.Second
+
+func billingSettlementRetryDelay(attempt int) time.Duration {
+	if attempt < 1 {
+		attempt = 1
+	}
+	delay := time.Second
+	for current := 1; current < attempt && delay < 15*time.Minute; current++ {
+		delay *= 2
+	}
+	if delay > 15*time.Minute {
+		return 15 * time.Minute
+	}
+	return delay
+}
+
+func billingSettlementErrorCode(err error) string {
+	if errors.Is(err, store.ErrInsufficientWalletBalance) {
+		return "insufficient_balance"
+	}
+	return "settlement_failed"
+}
+
+func billingSettlementErrorMessage(code string) string {
+	if code == "insufficient_balance" {
+		return "wallet balance is insufficient for settlement"
+	}
+	return "billing settlement processing failed"
+}
+
+func (s *Service) StartBillingSettlementWorker(ctx context.Context) {
+	workerID := "billing-" + uuid.NewString()
+	go s.runBillingSettlementWorker(ctx, workerID)
+}
+
+func (s *Service) runBillingSettlementWorker(ctx context.Context, workerID string) {
+	ticker := time.NewTicker(billingSettlementPollInterval)
+	defer ticker.Stop()
+
+	for {
+		s.processBillingSettlementBatch(ctx, workerID)
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+	}
+}
+
+func (s *Service) processBillingSettlementBatch(ctx context.Context, workerID string) {
+	items, err := s.store.ClaimBillingSettlements(ctx, workerID, store.BillingSettlementBatchSize, store.BillingSettlementLockTimeout)
+	if err != nil {
+		if ctx.Err() == nil {
+			s.logger.Error("claim billing settlements failed", "error_category", "billing_settlement_claim_failed")
+		}
+		return
+	}
+	for _, item := range items {
+		if ctx.Err() != nil {
+			return
+		}
+		if err := s.store.ProcessBillingSettlement(ctx, item); err != nil {
+			code := billingSettlementErrorCode(err)
+			markErr := s.store.MarkBillingSettlementFailed(
+				context.WithoutCancel(ctx),
+				item,
+				code,
+				billingSettlementErrorMessage(code),
+				billingSettlementRetryDelay(item.Attempts),
+			)
+			if markErr != nil {
+				s.logger.Error("mark billing settlement failed", "settlementID", item.ID, "taskID", item.TaskID, "error_category", "billing_settlement_state_failed")
+				continue
+			}
+			s.observeBillingEvent("settlement_retry")
+			if item.Attempts >= store.BillingSettlementMaxAttempts {
+				s.observeBillingEvent("manual_review")
+			}
+			s.logger.Warn("billing settlement scheduled for retry", "settlementID", item.ID, "taskID", item.TaskID, "action", item.Action, "error_category", code, "attempt", item.Attempts)
+			continue
+		}
+		s.observeBillingEvent("settlement_completed")
+		s.logger.Debug("billing settlement completed", "settlementID", item.ID, "taskID", item.TaskID, "action", item.Action)
+	}
+}

apps/api/internal/runner/billing_settlement_worker_test.go

@@ -0,0 +1,40 @@
+package runner
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
+)
+
+func TestBillingSettlementRetryDelay(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		attempt int
+		want    time.Duration
+	}{
+		{attempt: 1, want: time.Second},
+		{attempt: 2, want: 2 * time.Second},
+		{attempt: 10, want: 512 * time.Second},
+		{attempt: 11, want: 15 * time.Minute},
+		{attempt: 20, want: 15 * time.Minute},
+	}
+	for _, test := range tests {
+		if got := billingSettlementRetryDelay(test.attempt); got != test.want {
+			t.Fatalf("attempt %d: got %s, want %s", test.attempt, got, test.want)
+		}
+	}
+}
+
+func TestBillingSettlementErrorCode(t *testing.T) {
+	t.Parallel()
+
+	if got := billingSettlementErrorCode(store.ErrInsufficientWalletBalance); got != "insufficient_balance" {
+		t.Fatalf("got %q", got)
+	}
+	if got := billingSettlementErrorCode(errors.New("boom")); got != "settlement_failed" {
+		t.Fatalf("got %q", got)
+	}
+}

apps/api/internal/runner/execution_lease.go

@@ -0,0 +1,28 @@
+package runner
+
+import (
+	"context"
+	"time"
+)
+
+const (
+	taskExecutionLeaseTTL      = 5 * time.Minute
+	taskExecutionRenewInterval = 30 * time.Second
+)
+
+func (s *Service) renewTaskExecutionLease(ctx context.Context, cancel context.CancelFunc, taskID string, executionToken string) {
+	ticker := time.NewTicker(taskExecutionRenewInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if err := s.store.RenewTaskExecutionLease(ctx, taskID, executionToken, taskExecutionLeaseTTL); err != nil {
+				s.logger.Warn("task execution lease lost", "taskID", taskID, "error_category", "task_execution_lease_lost")
+				cancel()
+				return
+			}
+		}
+	}
+}

apps/api/internal/runner/pricing.go

@@ -37,19 +37,14 @@ func (s *Service) Estimate(ctx context.Context, kind string, model string, body
 		return EstimateResult{}, err
 	}
 	estimates := make([]candidateEstimate, 0, len(candidates))
-	var pricingErr error
 	for _, candidate := range candidates {
 		candidateBody := preprocessRequest(kind, cloneMap(body), candidate)
 		estimate, candidateErr := s.estimateCandidateV2(ctx, user, kind, candidateBody, candidate)
 		if candidateErr != nil {
-			pricingErr = candidateErr
-			continue
+			return EstimateResult{}, candidateErr
 		}
 		estimates = append(estimates, estimate)
 	}
-	if len(estimates) == 0 && pricingErr != nil {
-		return EstimateResult{}, pricingErr
-	}
 	return buildEstimateResult(estimates, pricingRequestFingerprint(kind, model, body))
 }
 

apps/api/internal/runner/queue_worker.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
 	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
+	"github.com/google/uuid"
 	"github.com/riverqueue/river"
 	"github.com/riverqueue/river/riverdriver/riverpgxv5"
 	"github.com/riverqueue/river/rivermigrate"
@@ -38,16 +39,22 @@ func (w *asyncTaskWorker) Work(ctx context.Context, job *river.Job[asyncTaskArgs
 	if task.Status == "succeeded" || task.Status == "failed" || task.Status == "cancelled" {
 		return nil
 	}
-	result, runErr := w.service.Execute(ctx, task, authUserFromTask(task))
+	executionToken := uuid.NewString()
+	result, runErr := w.service.executeWithToken(ctx, task, authUserFromTask(task), nil, executionToken)
 	if runErr == nil {
 		w.service.logger.Debug("river async task completed", "taskID", task.ID, "status", result.Task.Status, "riverJobID", job.ID)
 		return nil
 	}
+	if errors.Is(runErr, store.ErrTaskExecutionLeaseUnavailable) {
+		w.service.logger.Debug("river async task execution lease already held", "taskID", task.ID, "riverJobID", job.ID)
+		return nil
+	}
 	var queuedErr *TaskQueuedError
 	if errors.As(runErr, &queuedErr) {
 		return river.JobSnooze(queuedErr.Delay)
 	}
 	if ctx.Err() != nil {
+		task.ExecutionToken = executionToken
 		queued, queueErr := w.service.requeueInterruptedAsyncTask(context.WithoutCancel(ctx), task)
 		if queueErr != nil {
 			return queueErr
@@ -145,8 +152,7 @@ func (s *Service) recoverAsyncRiverJobs(ctx context.Context) error {
 		return err
 	}
 	for _, item := range items {
-		task := store.GatewayTask{ID: item.ID}
-		result, err := s.riverClient.Insert(ctx, asyncTaskArgs{TaskID: item.ID}, asyncTaskInsertOpts(task))
+		result, err := s.riverClient.Insert(ctx, asyncTaskArgs{TaskID: item.ID}, asyncTaskRecoveryInsertOpts(item))
 		if err != nil {
 			return err
 		}
@@ -186,6 +192,16 @@ func asyncTaskInsertOpts(task store.GatewayTask) *river.InsertOpts {
 	}
 }
 
+func asyncTaskRecoveryInsertOpts(item store.AsyncTaskQueueItem) *river.InsertOpts {
+	opts := asyncTaskInsertOpts(store.GatewayTask{ID: item.ID})
+	opts.ScheduledAt = item.NextRunAt
+	// A replacement process must not be blocked by a River row that the dead
+	// process left in running state. PostgreSQL execution leases still ensure
+	// that only one recovery job can call the upstream provider.
+	opts.UniqueOpts = river.UniqueOpts{}
+	return opts
+}
+
 func authUserFromTask(task store.GatewayTask) *auth.User {
 	roles := []string{"user"}
 	if strings.TrimSpace(task.UserID) == "" {

apps/api/internal/runner/service.go

@@ -14,6 +14,7 @@ import (
 	"github.com/easyai/easyai-ai-gateway/apps/api/internal/config"
 	scriptengine "github.com/easyai/easyai-ai-gateway/apps/api/internal/script"
 	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
+	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
 	"github.com/riverqueue/river"
 )
@@ -26,6 +27,11 @@ type Service struct {
 	scriptExecutor *scriptengine.Executor
 	httpClients    *httpClientCache
 	riverClient    *river.Client[pgx.Tx]
+	billingMetrics billingMetricsObserver
+}
+
+type billingMetricsObserver interface {
+	ObserveBillingEvent(string)
 }
 
 type Result struct {
@@ -39,6 +45,19 @@ type TaskQueuedError struct {
 	Delay time.Duration
 }
 
+type upstreamSubmissionUnknownError struct {
+	AttemptID string
+	Cause     error
+}
+
+func (e *upstreamSubmissionUnknownError) Error() string {
+	return "upstream submission result is unknown"
+}
+
+func (e *upstreamSubmissionUnknownError) Unwrap() error {
+	return e.Cause
+}
+
 func (e *TaskQueuedError) Error() string {
 	return ErrTaskQueued.Error()
 }
@@ -47,10 +66,10 @@ func (e *TaskQueuedError) Is(target error) bool {
 	return target == ErrTaskQueued
 }
 
-func New(cfg config.Config, db *store.Store, logger *slog.Logger) *Service {
+func New(cfg config.Config, db *store.Store, logger *slog.Logger, observers ...billingMetricsObserver) *Service {
 	httpClients := newHTTPClientCache()
 	scriptExecutor := &scriptengine.Executor{Logger: logger}
-	return &Service{
+	service := &Service{
 		cfg:            cfg,
 		store:          db,
 		logger:         logger,
@@ -76,6 +95,16 @@ func New(cfg config.Config, db *store.Store, logger *slog.Logger) *Service {
 		},
 		httpClients: httpClients,
 	}
+	if len(observers) > 0 {
+		service.billingMetrics = observers[0]
+	}
+	return service
+}
+
+func (s *Service) observeBillingEvent(event string) {
+	if s.billingMetrics != nil {
+		s.billingMetrics.ObserveBillingEvent(event)
+	}
 }
 
 func (s *Service) Execute(ctx context.Context, task store.GatewayTask, user *auth.User) (Result, error) {
@@ -87,6 +116,20 @@ func (s *Service) ExecuteStream(ctx context.Context, task store.GatewayTask, use
 }
 
 func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *auth.User, onDelta clients.StreamDelta) (Result, error) {
+	return s.executeWithToken(ctx, task, user, onDelta, uuid.NewString())
+}
+
+func (s *Service) executeWithToken(ctx context.Context, task store.GatewayTask, user *auth.User, onDelta clients.StreamDelta, executionToken string) (Result, error) {
+	wasRunning := task.Status == "running"
+	claimed, err := s.store.ClaimTaskExecution(ctx, task.ID, executionToken, taskExecutionLeaseTTL)
+	if err != nil {
+		return Result{}, err
+	}
+	task = claimed
+	executionCtx, stopExecution := context.WithCancel(ctx)
+	defer stopExecution()
+	go s.renewTaskExecutionLease(executionCtx, stopExecution, task.ID, task.ExecutionToken)
+	ctx = executionCtx
 	executeStartedAt := time.Now()
 	restoredRequest, err := s.restoreTaskRequestReferences(ctx, task)
 	if err != nil {
@@ -95,10 +138,10 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 	body := normalizeRequest(task.Kind, restoredRequest)
 	responseExecution := responseExecutionContext{}
 	modelType := modelTypeFromKind(task.Kind, body)
-	if err := s.store.MarkTaskRunning(ctx, task.ID, modelType, s.slimTaskRequestSnapshot(task, body)); err != nil {
+	if err := s.store.MarkTaskRunning(ctx, task.ID, task.ExecutionToken, modelType, s.slimTaskRequestSnapshot(task, body)); err != nil {
 		return Result{}, err
 	}
-	if task.Status != "running" {
+	if !wasRunning {
 		if err := s.emit(ctx, task.ID, "task.running", "running", "starting", 0.12, "task pulled from queue and started", map[string]any{"modelType": modelType}, task.RunMode == "simulation"); err != nil {
 			return Result{}, err
 		}
@@ -115,7 +158,7 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 			Reason:    "request_validation_failed",
 			ModelType: modelType,
 		})
-		failed, finishErr := s.failTask(ctx, task.ID, "bad_request", err.Error(), task.RunMode == "simulation", err)
+		failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, "bad_request", err.Error(), task.RunMode == "simulation", err)
 		if finishErr != nil {
 			return Result{}, finishErr
 		}
@@ -135,14 +178,14 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 			Reason:    "cloned_voice_binding_failed",
 			ModelType: modelType,
 		})
-		failed, finishErr := s.failTask(ctx, task.ID, clients.ErrorCode(err), err.Error(), task.RunMode == "simulation", err)
+		failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, clients.ErrorCode(err), err.Error(), task.RunMode == "simulation", err)
 		if finishErr != nil {
 			return Result{}, finishErr
 		}
 		return Result{Task: failed, Output: failed.Result}, err
 	}
 	if clonedVoice.Found {
-		if err := s.store.MarkTaskRunning(ctx, task.ID, modelType, s.slimTaskRequestSnapshot(task, body)); err != nil {
+		if err := s.store.MarkTaskRunning(ctx, task.ID, task.ExecutionToken, modelType, s.slimTaskRequestSnapshot(task, body)); err != nil {
 			return Result{}, err
 		}
 	}
@@ -151,7 +194,7 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 		if err != nil {
 			code, message := responseExecutionFailure(err)
 			s.recordFailedAttempt(ctx, failedAttemptRecord{Task: task, Body: body, AttemptNo: task.AttemptCount + 1, Code: code, Cause: err, Simulated: task.RunMode == "simulation", Scope: "response_chain", Reason: code, ModelType: modelType})
-			failed, finishErr := s.failTask(ctx, task.ID, code, message, task.RunMode == "simulation", err)
+			failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, code, message, task.RunMode == "simulation", err)
 			if finishErr != nil {
 				return Result{}, finishErr
 			}
@@ -183,7 +226,7 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 			Reason:    "candidate_selection_failed",
 			ModelType: modelType,
 		})
-		failed, finishErr := s.failTask(ctx, task.ID, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err)
+		failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err)
 		if finishErr != nil {
 			return Result{}, finishErr
 		}
@@ -202,7 +245,7 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 			Reason:    store.ModelCandidateErrorCode(err),
 			ModelType: modelType,
 		})
-		failed, finishErr := s.failTask(ctx, task.ID, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err)
+		failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err)
 		if finishErr != nil {
 			return Result{}, finishErr
 		}
@@ -224,7 +267,7 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 			ExtraMetrics: []map[string]any{candidateFilterMetrics},
 			ModelType:    modelType,
 		})
-		failed, finishErr := s.failTask(ctx, task.ID, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err, candidateFilterMetrics)
+		failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err, candidateFilterMetrics)
 		if finishErr != nil {
 			return Result{}, finishErr
 		}
@@ -240,7 +283,7 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 			Simulated: task.RunMode == "simulation", Scope: "candidate_output_token_filter", Reason: store.ModelCandidateErrorCode(err),
 			ExtraMetrics: []map[string]any{candidateFilterMetrics}, ModelType: modelType,
 		})
-		failed, finishErr := s.failTask(ctx, task.ID, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err, candidateFilterMetrics)
+		failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err, candidateFilterMetrics)
 		if finishErr != nil {
 			return Result{}, finishErr
 		}
@@ -251,7 +294,7 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 		if err != nil {
 			code, message := responseExecutionFailure(err)
 			s.recordFailedAttempt(ctx, failedAttemptRecord{Task: task, Body: body, AttemptNo: task.AttemptCount + 1, Code: code, Cause: err, Simulated: task.RunMode == "simulation", Scope: "response_protocol", Reason: code, ModelType: modelType})
-			failed, finishErr := s.failTask(ctx, task.ID, code, message, task.RunMode == "simulation", err)
+			failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, code, message, task.RunMode == "simulation", err)
 			if finishErr != nil {
 				return Result{}, finishErr
 			}
@@ -260,8 +303,17 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 	}
 	pricingByCandidate := map[string]resolvedPricing{}
 	reservationBillings := []any(nil)
+	reservationPricingSnapshot := map[string]any(nil)
 	if task.RunMode == "production" {
-		pricedCandidates := make([]store.RuntimeModelCandidate, 0, len(candidates))
+		billingMode := normalizedBillingEngineMode(s.cfg.BillingEngineMode)
+		if billingMode == "hold" {
+			holdErr := &clients.ClientError{Code: "billing_hold", Message: "production billing is temporarily on hold", Retryable: true}
+			failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, "billing_hold", holdErr.Error(), false, holdErr)
+			if finishErr != nil {
+				return Result{}, finishErr
+			}
+			return Result{Task: failed, Output: failed.Result}, holdErr
+		}
 		estimates := make([]candidateEstimate, 0, len(candidates))
 		var pricingErr error
 		for _, candidate := range candidates {
@@ -269,32 +321,68 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 			estimate, estimateErr := s.estimateCandidateV2(ctx, user, task.Kind, pricingBody, candidate)
 			if estimateErr != nil {
 				pricingErr = estimateErr
+				if billingMode == "enforce" {
+					break
+				}
 				continue
 			}
-			pricedCandidates = append(pricedCandidates, candidate)
 			estimates = append(estimates, estimate)
 			pricingByCandidate[pricingCandidateKey(candidate)] = estimate.Pricing
 		}
-		if len(pricedCandidates) == 0 {
-			if pricingErr == nil {
-				pricingErr = &PricingUnavailableError{Reason: "no candidate has effective pricing"}
+		if billingMode == "observe" {
+			legacyItems, legacyAmount := s.maximumLegacyCandidateEstimate(ctx, user, task.Kind, body, candidates)
+			reservationBillings = legacyItems
+			candidateSnapshots := make([]any, 0, len(estimates))
+			for _, estimate := range estimates {
+				candidateSnapshots = append(candidateSnapshots, estimate.Snapshot)
 			}
-			failed, finishErr := s.failTask(ctx, task.ID, "pricing_unavailable", pricingErr.Error(), false, pricingErr)
+			observedAmount := ""
+			if maximumEstimate, estimateErr := maximumCandidateEstimate(estimates); estimateErr == nil {
+				observedAmount = maximumEstimate.Amount.String()
+			}
+			reservationPricingSnapshot = map[string]any{
+				"pricingVersion": "legacy-observe", "observedPricingVersion": pricingVersionV2,
+				"requestFingerprint": pricingRequestFingerprint(task.Kind, task.Model, body),
+				"reservationAmount":  legacyAmount.String(), "observedReservationAmount": observedAmount,
+				"candidateCount": len(candidates), "observedCandidates": candidateSnapshots,
+			}
+			s.logger.Info("billing observe comparison", "taskID", task.ID, "legacyAmount", legacyAmount.String(), "v2Amount", observedAmount, "pricedCandidates", len(estimates), "candidateCount", len(candidates))
+		} else if pricingErr != nil || len(estimates) != len(candidates) {
+			if pricingErr == nil {
+				pricingErr = &PricingUnavailableError{Reason: "not every candidate has effective pricing"}
+			}
+			s.observeBillingEvent("pricing_unavailable")
+			failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, "pricing_unavailable", pricingErr.Error(), false, pricingErr)
 			if finishErr != nil {
 				return Result{}, finishErr
 			}
 			return Result{Task: failed, Output: failed.Result}, pricingErr
-		}
-		maximumEstimate, estimateErr := maximumCandidateEstimate(estimates)
-		if estimateErr != nil {
-			failed, finishErr := s.failTask(ctx, task.ID, "pricing_unavailable", estimateErr.Error(), false, estimateErr)
-			if finishErr != nil {
-				return Result{}, finishErr
+		} else {
+			maximumEstimate, estimateErr := maximumCandidateEstimate(estimates)
+			if estimateErr != nil {
+				failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, "pricing_unavailable", estimateErr.Error(), false, estimateErr)
+				if finishErr != nil {
+					return Result{}, finishErr
+				}
+				return Result{Task: failed, Output: failed.Result}, estimateErr
+			}
+			pricedCandidates := make([]store.RuntimeModelCandidate, 0, len(estimates))
+			candidateSnapshots := make([]any, 0, len(estimates))
+			for _, candidate := range candidates {
+				if _, ok := pricingByCandidate[pricingCandidateKey(candidate)]; ok {
+					pricedCandidates = append(pricedCandidates, candidate)
+				}
+			}
+			for _, estimate := range estimates {
+				candidateSnapshots = append(candidateSnapshots, estimate.Snapshot)
+			}
+			candidates = pricedCandidates
+			reservationBillings = maximumEstimate.Items
+			reservationPricingSnapshot = map[string]any{
+				"pricingVersion": pricingVersionV2, "requestFingerprint": pricingRequestFingerprint(task.Kind, task.Model, body),
+				"reservationAmount": maximumEstimate.Amount.String(), "candidateCount": len(estimates), "candidates": candidateSnapshots,
 			}
-			return Result{Task: failed, Output: failed.Result}, estimateErr
 		}
-		candidates = pricedCandidates
-		reservationBillings = maximumEstimate.Items
 	}
 	firstCandidateBody := body
 	normalizedModelType := modelType
@@ -328,17 +416,17 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 				Preprocessing: &firstPreprocessing,
 				ModelType:     normalizedModelType,
 			})
-			failed, finishErr := s.failTask(ctx, task.ID, clients.ErrorCode(clientErr), clientErr.Error(), task.RunMode == "simulation", clientErr, parameterPreprocessingMetrics(firstPreprocessing))
+			failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, clients.ErrorCode(clientErr), clientErr.Error(), task.RunMode == "simulation", clientErr, parameterPreprocessingMetrics(firstPreprocessing))
 			if finishErr != nil {
 				return Result{}, finishErr
 			}
 			return Result{Task: failed, Output: failed.Result}, clientErr
 		}
-		if err := s.store.MarkTaskRunning(ctx, task.ID, candidates[0].ModelType, s.slimTaskRequestSnapshot(task, firstCandidateBody)); err != nil {
+		if err := s.store.MarkTaskRunning(ctx, task.ID, task.ExecutionToken, candidates[0].ModelType, s.slimTaskRequestSnapshot(task, firstCandidateBody)); err != nil {
 			return Result{}, err
 		}
 		var reserveErr error
-		walletReservations, reserveErr = s.store.ReserveTaskBilling(ctx, task, user, reservationBillings)
+		walletReservations, reserveErr = s.store.ReserveTaskBilling(ctx, task, user, reservationBillings, reservationPricingSnapshot)
 		if reserveErr != nil {
 			if errors.Is(reserveErr, store.ErrInsufficientWalletBalance) {
 				attemptNo = s.recordFailedAttempt(ctx, failedAttemptRecord{
@@ -355,7 +443,7 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 					Preprocessing: &firstPreprocessing,
 					ModelType:     normalizedModelType,
 				})
-				failed, finishErr := s.failTask(ctx, task.ID, "insufficient_balance", reserveErr.Error(), task.RunMode == "simulation", reserveErr, parameterPreprocessingMetrics(firstPreprocessing))
+				failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, "insufficient_balance", reserveErr.Error(), task.RunMode == "simulation", reserveErr, parameterPreprocessingMetrics(firstPreprocessing))
 				if finishErr != nil {
 					return Result{}, finishErr
 				}
@@ -406,19 +494,46 @@ candidatesLoop:
 				break candidatesLoop
 			}
 			candidateBody := preprocessing.Body
-			response, err := s.runCandidate(ctx, task, user, candidateBody, preprocessing.Log, candidate, nextAttemptNo, onDelta, responseExecution, singleSourceProtected, runnerPolicy.CacheAffinityPolicy, cacheAffinityKeys.Record)
+			candidatePricing := pricingByCandidate[pricingCandidateKey(candidate)]
+			response, err := s.runCandidate(ctx, task, user, candidateBody, preprocessing.Log, candidate, candidatePricing, nextAttemptNo, onDelta, responseExecution, singleSourceProtected, runnerPolicy.CacheAffinityPolicy, cacheAffinityKeys.Record)
 			if err == nil {
 				attemptNo = nextAttemptNo
 				var billings []any
+				finalAmount := fixedAmount(0)
+				finalAmountText := ""
+				pricingSnapshot := map[string]any{"pricingVersion": pricingVersionV2, "source": "simulation"}
 				if task.RunMode == "production" {
-					pricing := pricingByCandidate[pricingCandidateKey(candidate)]
+					billingMode := normalizedBillingEngineMode(s.cfg.BillingEngineMode)
 					var billingErr error
-					billings, _, _, billingErr = s.billingsWithResolvedPricingV2(ctx, user, task.Kind, candidateBody, candidate, response, false, pricing)
-					if billingErr != nil {
-						// The upstream result may already exist. Preserve the reservation for manual review.
-						walletReservationFinalized = true
-						return Result{}, billingErr
+					if billingMode == "observe" {
+						billings = s.billings(ctx, user, task.Kind, candidateBody, candidate, response, false)
+						finalAmount = billingItemsFixedTotal(billings)
+						pricingSnapshot = map[string]any{
+							"pricingVersion": "legacy-observe", "observedPricingVersion": pricingVersionV2,
+							"observedPricing": candidatePricing.Snapshot,
+						}
+					} else {
+						billings, finalAmount, _, billingErr = s.billingsWithResolvedPricingV2(ctx, user, task.Kind, candidateBody, candidate, response, false, candidatePricing)
+						pricingSnapshot = candidatePricing.Snapshot
 					}
+					if billingErr != nil {
+						review, reviewErr := s.store.FinishTaskManualReview(context.WithoutCancel(ctx), store.FinishTaskManualReviewInput{
+							TaskID: task.ID, ExecutionToken: task.ExecutionToken, AttemptID: response.AttemptID, TaskStatus: "succeeded",
+							Code: "billing_calculation_failed", Message: "billing calculation requires manual review",
+							Result: response.Result, RequestID: response.RequestID,
+							PricingSnapshot:    candidatePricing.Snapshot,
+							RequestFingerprint: pricingRequestFingerprint(task.Kind, task.Model, candidateBody),
+							ResponseStartedAt:  response.ResponseStartedAt, ResponseFinishedAt: response.ResponseFinishedAt,
+							ResponseDurationMS: response.ResponseDurationMS,
+						})
+						if reviewErr != nil {
+							return Result{}, reviewErr
+						}
+						walletReservationFinalized = true
+						s.logger.Warn("task succeeded but billing requires manual review", "taskID", task.ID, "error_category", "billing_calculation_failed")
+						return Result{Task: review, Output: response.Result}, nil
+					}
+					finalAmountText = finalAmount.String()
 				} else {
 					billings = s.billings(ctx, user, task.Kind, candidateBody, candidate, response, true)
 				}
@@ -427,38 +542,32 @@ candidatesLoop:
 				record.Metrics = mergeMetrics(record.Metrics, parameterPreprocessingMetrics(preprocessing.Log))
 				record.Metrics = s.withAttemptHistory(ctx, task.ID, record.Metrics)
 				finished, finishErr := s.store.FinishTaskSuccess(ctx, store.FinishTaskSuccessInput{
-					TaskID:             task.ID,
-					Result:             response.Result,
-					Billings:           billings,
-					RequestID:          record.RequestID,
-					ResolvedModel:      record.ResolvedModel,
-					Usage:              record.Usage,
-					Metrics:            record.Metrics,
-					BillingSummary:     record.BillingSummary,
-					FinalChargeAmount:  record.FinalChargeAmount,
-					ResponseStartedAt:  record.ResponseStartedAt,
-					ResponseFinishedAt: record.ResponseFinishedAt,
-					ResponseDurationMS: record.ResponseDurationMS,
+					TaskID:                task.ID,
+					ExecutionToken:        task.ExecutionToken,
+					AttemptID:             response.AttemptID,
+					Result:                response.Result,
+					Billings:              billings,
+					RequestID:             record.RequestID,
+					ResolvedModel:         record.ResolvedModel,
+					Usage:                 record.Usage,
+					Metrics:               record.Metrics,
+					BillingSummary:        record.BillingSummary,
+					FinalChargeAmount:     record.FinalChargeAmount,
+					FinalChargeAmountText: finalAmountText,
+					BillingCurrency:       stringFromAny(record.BillingSummary["currency"]),
+					PricingSnapshot:       pricingSnapshot,
+					RequestFingerprint:    pricingRequestFingerprint(task.Kind, task.Model, candidateBody),
+					ResponseStartedAt:     record.ResponseStartedAt,
+					ResponseFinishedAt:    record.ResponseFinishedAt,
+					ResponseDurationMS:    record.ResponseDurationMS,
 				})
 				if finishErr != nil {
 					return Result{}, finishErr
 				}
-				if finished.FinalChargeAmount > 0 {
-					walletReservationFinalized = true
-					if settleErr := s.store.SettleTaskBilling(ctx, finished); settleErr != nil {
-						return Result{}, settleErr
-					}
-				} else if len(walletReservations) > 0 {
-					if releaseErr := s.store.ReleaseTaskBillingReservations(ctx, walletReservations, "task_billing_zero"); releaseErr != nil {
-						return Result{}, releaseErr
-					}
-					walletReservationFinalized = true
-				}
 				walletReservationFinalized = true
-				if finished.FinalChargeAmount > 0 {
-					if err := s.emit(ctx, task.ID, "task.billing.settled", "succeeded", "billing", 0.98, "task billing settled", map[string]any{
-						"amount":   finished.FinalChargeAmount,
-						"currency": stringFromAny(record.BillingSummary["currency"]),
+				if finished.BillingStatus == "pending" {
+					if err := s.emit(ctx, task.ID, "task.billing.pending", "succeeded", "billing", 0.98, "task billing queued", map[string]any{
+						"amount": finished.FinalChargeAmount, "currency": finished.BillingCurrency,
 					}, isSimulation(task, candidate)); err != nil {
 						return Result{}, err
 					}
@@ -475,6 +584,21 @@ candidatesLoop:
 				}
 				return Result{Task: finished, Output: response.Result}, nil
 			}
+			var submissionUnknown *upstreamSubmissionUnknownError
+			if errors.As(err, &submissionUnknown) {
+				review, reviewErr := s.store.FinishTaskManualReview(context.WithoutCancel(ctx), store.FinishTaskManualReviewInput{
+					TaskID: task.ID, ExecutionToken: task.ExecutionToken, AttemptID: submissionUnknown.AttemptID, TaskStatus: "failed",
+					Code: "upstream_submission_unknown", Message: submissionUnknown.Error(),
+					PricingSnapshot:    candidatePricing.Snapshot,
+					RequestFingerprint: pricingRequestFingerprint(task.Kind, task.Model, candidateBody),
+				})
+				if reviewErr != nil {
+					return Result{}, reviewErr
+				}
+				walletReservationFinalized = true
+				s.logger.Warn("upstream submission requires manual review", "taskID", task.ID, "attemptID", submissionUnknown.AttemptID, "error_category", "upstream_submission_unknown")
+				return Result{Task: review, Output: review.Result}, submissionUnknown
+			}
 			if isLocalRateLimitError(err) {
 				lastErr = err
 				candidateErr = err
@@ -622,10 +746,12 @@ candidatesLoop:
 	if lastPreprocessing != nil {
 		extraMetrics = append(extraMetrics, parameterPreprocessingMetrics(*lastPreprocessing))
 	}
-	failed, err := s.failTask(ctx, task.ID, code, message, task.RunMode == "simulation", lastErr, extraMetrics...)
+	failed, err := s.failTask(ctx, task.ID, task.ExecutionToken, code, message, task.RunMode == "simulation", lastErr, extraMetrics...)
 	if err != nil {
 		return Result{}, err
 	}
+	// FinishTaskFailure atomically transfers ownership of any reservation to the release Outbox.
+	walletReservationFinalized = true
 	return Result{Task: failed, Output: failed.Result}, lastErr
 }
 
@@ -633,7 +759,46 @@ func pricingCandidateKey(candidate store.RuntimeModelCandidate) string {
 	return firstNonEmptyString(candidate.PlatformModelID, candidate.PlatformID+":"+candidate.ModelName)
 }
 
-func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user *auth.User, body map[string]any, preprocessing parameterPreprocessingLog, candidate store.RuntimeModelCandidate, attemptNo int, onDelta clients.StreamDelta, responseExecution responseExecutionContext, singleSourceProtected bool, cacheAffinityPolicy map[string]any, cacheAffinityRecordKeys []string) (clients.Response, error) {
+func normalizedBillingEngineMode(value string) string {
+	switch strings.ToLower(strings.TrimSpace(value)) {
+	case "enforce", "hold":
+		return strings.ToLower(strings.TrimSpace(value))
+	default:
+		return "observe"
+	}
+}
+
+func (s *Service) maximumLegacyCandidateEstimate(ctx context.Context, user *auth.User, kind string, body map[string]any, candidates []store.RuntimeModelCandidate) ([]any, fixedAmount) {
+	var maximumItems []any
+	maximumAmount := fixedAmount(0)
+	for index, candidate := range candidates {
+		candidateBody := preprocessRequest(kind, cloneMap(body), candidate)
+		items := s.estimatedBillings(ctx, user, kind, candidateBody, candidate)
+		amount := billingItemsFixedTotal(items)
+		if index == 0 || amount > maximumAmount {
+			maximumItems = items
+			maximumAmount = amount
+		}
+	}
+	return maximumItems, maximumAmount
+}
+
+func billingItemsFixedTotal(items []any) fixedAmount {
+	total := fixedAmount(0)
+	for _, raw := range items {
+		line, _ := raw.(map[string]any)
+		if line == nil {
+			continue
+		}
+		amount, err := fixedAmountFromAny(line["amount"])
+		if err == nil && amount > 0 {
+			total = total.Add(amount)
+		}
+	}
+	return total
+}
+
+func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user *auth.User, body map[string]any, preprocessing parameterPreprocessingLog, candidate store.RuntimeModelCandidate, pricing resolvedPricing, attemptNo int, onDelta clients.StreamDelta, responseExecution responseExecutionContext, singleSourceProtected bool, cacheAffinityPolicy map[string]any, cacheAffinityRecordKeys []string) (clients.Response, error) {
 	simulated := isSimulation(task, candidate)
 	baseAttemptMetrics := mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), parameterPreprocessingMetrics(preprocessing))
 	reservations := s.rateLimitReservations(ctx, user, candidate, body)
@@ -652,16 +817,18 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 	defer s.store.ReleaseConcurrencyLeases(context.WithoutCancel(ctx), limitResult.LeaseIDs)
 
 	attemptID, err := s.store.CreateTaskAttempt(ctx, store.CreateTaskAttemptInput{
-		TaskID:          task.ID,
-		AttemptNo:       attemptNo,
-		PlatformID:      candidate.PlatformID,
-		PlatformModelID: candidate.PlatformModelID,
-		ClientID:        candidate.ClientID,
-		QueueKey:        candidate.QueueKey,
-		Status:          "running",
-		Simulated:       simulated,
-		RequestSnapshot: s.slimTaskRequestSnapshot(task, body),
-		Metrics:         baseAttemptMetrics,
+		TaskID:             task.ID,
+		AttemptNo:          attemptNo,
+		PlatformID:         candidate.PlatformID,
+		PlatformModelID:    candidate.PlatformModelID,
+		ClientID:           candidate.ClientID,
+		QueueKey:           candidate.QueueKey,
+		Status:             "running",
+		Simulated:          simulated,
+		RequestSnapshot:    s.slimTaskRequestSnapshot(task, body),
+		Metrics:            baseAttemptMetrics,
+		PricingSnapshot:    pricing.Snapshot,
+		RequestFingerprint: pricingRequestFingerprint(task.Kind, task.Model, body),
 	})
 	if err != nil {
 		return clients.Response{}, fmt.Errorf("create task attempt: %w", err)
@@ -731,6 +898,9 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 		publicResponseID = responseExecution.PublicResponseID
 		publicPreviousResponseID = responseExecution.PublicPreviousResponseID
 	}
+	if err := s.store.SetAttemptUpstreamSubmissionStatus(ctx, attemptID, "submitting"); err != nil {
+		return clients.Response{}, fmt.Errorf("mark upstream submission: %w", err)
+	}
 	response, err := client.Run(ctx, clients.Request{
 		Kind:              task.Kind,
 		ModelType:         candidate.ModelType,
@@ -744,7 +914,7 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 			if strings.TrimSpace(remoteTaskID) == "" {
 				return nil
 			}
-			return s.store.SetTaskRemoteTask(context.WithoutCancel(ctx), task.ID, attemptID, remoteTaskID, payload)
+			return s.store.SetTaskRemoteTask(context.WithoutCancel(ctx), task.ID, task.ExecutionToken, attemptID, remoteTaskID, payload)
 		},
 		Stream:                     boolFromMap(providerBody, "stream"),
 		StreamDelta:                onDelta,
@@ -768,6 +938,9 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 		}
 	}
 	if err != nil {
+		if clients.ErrorResponseMetadata(err).StatusCode > 0 {
+			_ = s.store.SetAttemptUpstreamSubmissionStatus(context.WithoutCancel(ctx), attemptID, "response_received")
+		}
 		retryable := clients.IsRetryable(err)
 		requestID, metrics, responseStartedAt, responseFinishedAt, responseDurationMS := failureMetrics(err, simulated)
 		if responseStartedAt.IsZero() {
@@ -796,6 +969,9 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 			ErrorMessage:       err.Error(),
 		})
 		_ = s.emit(ctx, task.ID, "task.attempt.failed", "running", "attempt_failed", 0.45, err.Error(), map[string]any{"attempt": attemptNo, "retryable": retryable, "requestId": requestID, "statusCode": clients.ErrorResponseMetadata(err).StatusCode, "metrics": metrics}, simulated)
+		if !simulated && clients.ErrorResponseMetadata(err).StatusCode == 0 {
+			return clients.Response{}, &upstreamSubmissionUnknownError{AttemptID: attemptID, Cause: err}
+		}
 		s.applyCandidateFailurePolicies(ctx, task.ID, candidate, err, simulated, singleSourceProtected)
 		return clients.Response{}, err
 	}
@@ -877,19 +1053,7 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 		return clients.Response{}, fmt.Errorf("commit rate limit reservations: %w", err)
 	}
 	rateReservationsFinalized = true
-	if err := s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
-		AttemptID:          attemptID,
-		Status:             "succeeded",
-		RequestID:          response.RequestID,
-		Usage:              usageToMap(response.Usage),
-		Metrics:            mergeMetrics(taskMetrics(task, user, body, candidate, response, simulated), parameterPreprocessingMetrics(preprocessing)),
-		ResponseSnapshot:   response.Result,
-		ResponseStartedAt:  response.ResponseStartedAt,
-		ResponseFinishedAt: response.ResponseFinishedAt,
-		ResponseDurationMS: response.ResponseDurationMS,
-	}); err != nil {
-		return clients.Response{}, fmt.Errorf("finish task attempt: %w", err)
-	}
+	response.AttemptID = attemptID
 	if err := s.store.RecordCacheAffinityObservation(context.WithoutCancel(ctx), store.CacheAffinityObservationInput{
 		CacheAffinityKey:       candidate.CacheAffinity.Key,
 		CacheAffinityKeys:      cacheAffinityRecordKeys,
@@ -957,7 +1121,7 @@ func (s *Service) clientFor(candidate store.RuntimeModelCandidate, simulated boo
 	return s.clients["openai"]
 }
 
-func (s *Service) failTask(ctx context.Context, taskID string, code string, message string, simulated bool, cause error, extraMetrics ...map[string]any) (store.GatewayTask, error) {
+func (s *Service) failTask(ctx context.Context, taskID string, executionToken string, code string, message string, simulated bool, cause error, extraMetrics ...map[string]any) (store.GatewayTask, error) {
 	requestID, metrics, responseStartedAt, responseFinishedAt, responseDurationMS := failureMetrics(cause, simulated)
 	if len(extraMetrics) > 0 {
 		values := append([]map[string]any{metrics}, extraMetrics...)
@@ -966,6 +1130,7 @@ func (s *Service) failTask(ctx context.Context, taskID string, code string, mess
 	metrics = s.withAttemptHistory(ctx, taskID, metrics)
 	failed, err := s.store.FinishTaskFailure(ctx, store.FinishTaskFailureInput{
 		TaskID:             taskID,
+		ExecutionToken:     executionToken,
 		Code:               code,
 		Message:            message,
 		Result:             buildFailureResult(code, message, requestID, cause),
@@ -1094,7 +1259,7 @@ func (s *Service) requeueRateLimitedTask(ctx context.Context, task store.Gateway
 	if delay <= 0 {
 		delay = 5 * time.Second
 	}
-	queued, err := s.store.RequeueTask(ctx, task.ID, delay, candidate.QueueKey)
+	queued, err := s.store.RequeueTask(ctx, task.ID, task.ExecutionToken, delay, candidate.QueueKey)
 	if err != nil {
 		return store.GatewayTask{}, 0, err
 	}
@@ -1110,7 +1275,7 @@ func (s *Service) requeueRateLimitedTask(ctx context.Context, task store.Gateway
 }
 
 func (s *Service) requeueInterruptedAsyncTask(ctx context.Context, task store.GatewayTask) (store.GatewayTask, error) {
-	queued, err := s.store.RequeueTask(ctx, task.ID, 0, "")
+	queued, err := s.store.RequeueTask(ctx, task.ID, task.ExecutionToken, 0, "")
 	if err != nil {
 		return store.GatewayTask{}, err
 	}

apps/api/internal/runner/wallet_execute_test.go

@@ -182,6 +182,20 @@ func TestExecuteWithMockClientRejectsConcurrentTasksBeyondWalletBalance(t *testi
 	if got := mockClient.calls.Load(); got != 1 {
 		t.Fatalf("mock client calls = %d, want 1", got)
 	}
+	settlements, err := db.ClaimBillingSettlements(ctx, "wallet-execute-test", store.BillingSettlementBatchSize, store.BillingSettlementLockTimeout)
+	if err != nil {
+		t.Fatalf("claim billing settlements: %v", err)
+	}
+	processed := 0
+	for _, settlement := range settlements {
+		if err := db.ProcessBillingSettlement(ctx, settlement); err != nil {
+			t.Fatalf("process billing settlement %s: %v", settlement.ID, err)
+		}
+		processed++
+	}
+	if processed != 1 {
+		t.Fatalf("processed billing settlements = %d, want 1", processed)
+	}
 
 	summary, err := db.GetWalletSummary(ctx, user, "resource")
 	if err != nil {