feat: notification Runner + multi-token Cost; simulation cogs fix (#530)

Co-authored-by: Will Zimmerman <will@avaprotocol.org>

Author: chrisli30

aggregator/rpc_server.go

@@ -55,49 +55,6 @@ type RpcServer struct {
 	chainID          *big.Int
 }
 
-// FallbackPriceService provides hardcoded fallback prices when Moralis is unavailable
-type FallbackPriceService struct{}
-
-func newFallbackPriceService() *FallbackPriceService {
-	return &FallbackPriceService{}
-}
-
-func (fps *FallbackPriceService) GetNativeTokenPriceUSD(chainID int64) (*big.Float, error) {
-	// Only include chains that the aggregator actually supports: Ethereum and Base
-	fallbackPrices := map[int64]float64{
-		1:        2500.0, // Ethereum Mainnet
-		11155111: 2500.0, // Ethereum Sepolia
-		8453:     2500.0, // Base Mainnet
-		84532:    2500.0, // Base Sepolia
-	}
-
-	if price, exists := fallbackPrices[chainID]; exists {
-		return big.NewFloat(price), nil
-	}
-	return big.NewFloat(2500.0), nil // Default ETH price
-}
-
-func (fps *FallbackPriceService) GetNativeTokenSymbol(chainID int64) string {
-	// Only include chains that the aggregator actually supports: Ethereum and Base
-	// All supported chains use ETH as the native token
-	tokenSymbols := map[int64]string{
-		1:        "ETH", // Ethereum Mainnet
-		11155111: "ETH", // Ethereum Sepolia
-		8453:     "ETH", // Base Mainnet
-		84532:    "ETH", // Base Sepolia
-	}
-
-	if symbol, exists := tokenSymbols[chainID]; exists {
-		return symbol
-	}
-	return "ETH"
-}
-
-// FallbackPriceInfo provides information about fallback pricing for logging
-func (fps *FallbackPriceService) FallbackPriceInfo() string {
-	return "using conservative ETH price of $2500"
-}
-
 // Get nonce of an existing smart wallet of a given owner
 func (r *RpcServer) GetWallet(ctx context.Context, payload *avsproto.GetWalletReq) (*avsproto.GetWalletResp, error) {
 	user, err := r.verifyAuth(ctx)
@@ -1191,23 +1148,15 @@ func (r *RpcServer) EstimateFees(ctx context.Context, req *avsproto.EstimateFees
 		return nil, status.Errorf(codes.InvalidArgument, "expire_at must be after created_at")
 	}
 
-	// Create price service (Moralis if API key available, otherwise fallback)
+	// Price service is required for USD-equivalent fee numbers. When Moralis
+	// isn't configured, callers receive cogs (WEI) and executionFee (USD) as
+	// raw values without USD-equivalent conversions — and notifications render
+	// "$?" rather than a fabricated number.
 	var priceService taskengine.PriceService
 	if r.config.MoralisApiKey != "" {
 		priceService = services.GetMoralisService(r.config.MoralisApiKey, r.config.Logger)
 	} else {
-		priceService = newFallbackPriceService()
-		var fallbackPriceInfo string
-		// Try to extract fallback price info for logging
-		type fallbackPricer interface {
-			FallbackPriceInfo() string
-		}
-		if fp, ok := priceService.(fallbackPricer); ok {
-			fallbackPriceInfo = fp.FallbackPriceInfo()
-		} else {
-			fallbackPriceInfo = "unknown fallback price"
-		}
-		r.config.Logger.Warn(fmt.Sprintf("No Moralis API key configured, using fallback price service for fee estimation (%s)", fallbackPriceInfo))
+		r.config.Logger.Warn("No Moralis API key configured; fee estimates will lack USD-equivalent conversions and notifications will render $? for token totals")
 	}
 
 	// Create fee estimator - use configuration-aware version if fee rates are configured

aggregator/task_engine.go

@@ -97,15 +97,17 @@ func (agg *Aggregator) startTaskEngine(ctx context.Context) {
 		agg.logger,
 	)
 
-	// Create price service for fee conversion (USD → ETH)
+	// Price service for fee conversion (USD → ETH and ERC20 lookups). When
+	// Moralis isn't configured, it stays nil and callers gracefully degrade
+	// (notifications render "$?" for unknown prices).
 	var priceService taskengine.PriceService
 	if agg.config.MoralisApiKey != "" {
 		priceService = services.GetMoralisService(agg.config.MoralisApiKey, agg.logger)
 	} else {
-		priceService = newFallbackPriceService()
+		agg.logger.Warn("No Moralis API key configured; USD-equivalent fee numbers will be unavailable")
 	}
 
-	// Store price service on engine for use in simulation path
+	// Store price service on engine (nil-safe — engine and summarizer handle absence).
 	agg.engine.SetPriceService(priceService)
 
 	// Create executor with engine reference for atomic execution indexing

core/services/moralis_service.go

@@ -193,6 +193,53 @@ func (ms *MoralisService) GetNativeTokenSymbol(chainID int64) string {
 	return "ETH" // Default fallback
 }
 
+// GetERC20PriceUSD fetches the USD price for an ERC20 contract on the given
+// chain. Returns an error when the price isn't available — callers render
+// "$?" rather than fabricate a number. No internal $1.00 / fallback.
+//
+// NOTE: Stablecoin shortcut (taskengine.LookupStablecoinSymbol → $1.00) is
+// applied at the caller layer in taskengine, not here, to keep the
+// services/taskengine dependency direction correct.
+func (ms *MoralisService) GetERC20PriceUSD(chainID int64, contractAddress string) (*big.Float, error) {
+	if contractAddress == "" {
+		return nil, fmt.Errorf("contract address required")
+	}
+
+	moralisChain := ms.chainIDToMoralisChain(chainID)
+	if moralisChain == "" {
+		return nil, fmt.Errorf("unsupported chain ID: %d", chainID)
+	}
+
+	cacheKey := fmt.Sprintf("erc20_%d_%s", chainID, contractAddress)
+	if cached := ms.getCachedPrice(cacheKey); cached != nil {
+		return cached.Price, nil
+	}
+
+	if ms.apiKey == "" {
+		return nil, fmt.Errorf("moralis API key not configured")
+	}
+
+	url := fmt.Sprintf("https://deep-index.moralis.io/api/v2.2/erc20/%s/price", contractAddress)
+	resp, err := ms.httpClient.R().
+		SetQueryParams(map[string]string{"chain": moralisChain}).
+		SetHeader("X-API-Key", ms.apiKey).
+		SetResult(&MoralisTokenPriceResponse{}).
+		Get(url)
+	if err != nil {
+		return nil, fmt.Errorf("moralis ERC20 price request failed: %w", err)
+	}
+	if resp.StatusCode() != 200 {
+		return nil, fmt.Errorf("moralis ERC20 price returned status %d: %s", resp.StatusCode(), resp.String())
+	}
+	result := resp.Result().(*MoralisTokenPriceResponse)
+	if result.UsdPrice <= 0 {
+		return nil, fmt.Errorf("invalid ERC20 price from Moralis: %f", result.UsdPrice)
+	}
+	price := big.NewFloat(result.UsdPrice)
+	ms.setCachedPrice(cacheKey, price, "")
+	return price, nil
+}
+
 // GetPriceDataAge returns the age of cached price data in seconds
 func (ms *MoralisService) GetPriceDataAge(chainID int64) int64 {
 	cacheKey := fmt.Sprintf("chain_%d", chainID)

core/taskengine/blockchain_constants.go

@@ -1,5 +1,7 @@
 package taskengine
 
+import "strings"
+
 // Ethereum gas cost constants
 const (
 	// StandardGasCost represents the standard gas cost for a simple Ethereum transaction (21000 gas)
@@ -25,6 +27,94 @@ const (
 	DefaultGasPriceHex = "0x1dcd6500"
 )
 
+// DefaultGasPriceByChain provides per-chain conservative simulation gas prices (wei).
+// Chosen to lean slightly high so simulated cogs don't undershoot real costs.
+// Real-time gas pricing is a separate follow-up.
+var DefaultGasPriceByChain = map[uint64]uint64{
+	1:        5_000_000_000, // Ethereum Mainnet — 5 gwei
+	11155111: 500_000_000,   // Ethereum Sepolia — 0.5 gwei
+	8453:     50_000_000,    // Base — 0.05 gwei
+	84532:    10_000_000,    // Base Sepolia — 0.01 gwei
+}
+
+// GetDefaultGasPrice returns the per-chain default simulation gas price (wei),
+// falling back to DefaultGasPrice (0.5 gwei) for unknown chains.
+func GetDefaultGasPrice(chainID uint64) uint64 {
+	if v, ok := DefaultGasPriceByChain[chainID]; ok {
+		return v
+	}
+	return DefaultGasPrice
+}
+
+// StablecoinInfo carries the display symbol and ERC20 decimals for a stablecoin
+// hard-coded as $1.00. Decimals are required to format raw token amounts.
+type StablecoinInfo struct {
+	Symbol   string
+	Decimals uint32
+}
+
+// Stablecoins maps chain ID → lowercased contract address → StablecoinInfo
+// for fully-reserved or strongly-collateralized USD stablecoins. Lookups
+// treat each listed address as exactly $1.00 USD without a price-service
+// network hop — covers the bulk of real-world value-fee calculation cases.
+// Tokens not in this map fall through to PriceService.GetERC20PriceUSD; on
+// miss the renderer prints the "$?" placeholder.
+//
+// Inclusion criteria (high bar — incorrect ≈$1.00 assumptions miscompute fees):
+//   - Fully reserved by audited issuer (Circle, Paxos, Tether, PayPal, Ripple,
+//     First Digital, TrueUSD, Gemini), OR
+//   - Overcollateralized by crypto with strong peg history (DAI, USDS, LUSD,
+//     sDAI which redeems 1:1 against DAI).
+//
+// Algorithmic / synthetic / new-untested stablecoins (USDe, USDD, USD1, USDF,
+// FRAX) are deliberately excluded — they go through the price service like any
+// other ERC20 so a depeg event surfaces correctly.
+var Stablecoins = map[uint64]map[string]StablecoinInfo{
+	// Ethereum Mainnet
+	1: {
+		"0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48": {"USDC", 6},
+		"0xdac17f958d2ee523a2206206994597c13d831ec7": {"USDT", 6},
+		"0x6b175474e89094c44da98b954eedeac495271d0f": {"DAI", 18},
+		"0xdc035d45d973e3ec169d2276ddab16f1e407384f": {"USDS", 18},
+		"0x6c3ea9036406852006290770bedfcaba0e23a0e8": {"PYUSD", 6},
+		"0x83f20f44975d03b1b09e64809b757c47f942beea": {"sDAI", 18},
+		"0xc5f0f7b66764f6ec8c8dff7ba683102295e16409": {"FDUSD", 18},
+		"0x0000000000085d4780b73119b644ae5ecd22b376": {"TUSD", 18},
+		"0x056fd409e1d7a124bd7017459dfea2f387b6d5cd": {"GUSD", 2},
+		"0x5f98805a4e8be255a32880fdec7f6728c6568ba0": {"LUSD", 18},
+		"0x8292bb45bf1ee4d140127049757c2e0ff06317ed": {"RLUSD", 18},
+		"0xe343167631d89b6ffc58b88d6b7fb0228795491d": {"USDG", 6},
+	},
+	// Base Mainnet
+	8453: {
+		"0x833589fcd6edb6e08f4c7c32d4f71b54bda02913": {"USDC", 6}, // Circle native
+		"0xd9aaec86b65d86f6a7b5b1b0c42ffa531710b6ca": {"USDC", 6}, // bridged (legacy)
+		"0xfde4c96c8593536e31f229ea8f37b2ada2699bb2": {"USDT", 6},
+		"0x50c5725949a6f0c72e6c4a641f24049a917db0cb": {"DAI", 18},
+	},
+	// Ethereum Sepolia (testnet)
+	11155111: {
+		"0x1c7d4b196cb0c7b01d743fbc6116a902379c7238": {"USDC", 6}, // Circle test deployment
+		"0xaa8e23fb1079ea71e0a56f48a2aa51851d8433d0": {"USDT", 6},
+	},
+	// Base Sepolia (testnet)
+	84532: {
+		"0x036cbd53842c5426634e7929541ec2318f3dcf7e": {"USDC", 6}, // Circle test deployment
+	},
+}
+
+// LookupStablecoin returns symbol+decimals for a stablecoin contract, or
+// (StablecoinInfo{}, false) if the address isn't in the chain's hard-coded
+// $1.00 list. Address matching is case-insensitive.
+func LookupStablecoin(chainID uint64, contractAddress string) (StablecoinInfo, bool) {
+	chainMap, ok := Stablecoins[chainID]
+	if !ok {
+		return StablecoinInfo{}, false
+	}
+	info, ok := chainMap[strings.ToLower(contractAddress)]
+	return info, ok
+}
+
 // Contract method constants
 const (
 	// UnknownMethodName represents a placeholder for contract method names that need to be resolved from ABI

core/taskengine/blockchain_constants_test.go

@@ -150,3 +150,60 @@ func BenchmarkGetChainSearchRanges(b *testing.B) {
 		_ = GetChainSearchRanges(chainID)
 	}
 }
+
+func TestGetDefaultGasPrice(t *testing.T) {
+	cases := []struct {
+		name    string
+		chainID uint64
+		want    uint64
+	}{
+		{"Ethereum Mainnet", 1, 5_000_000_000},
+		{"Ethereum Sepolia", 11155111, 500_000_000},
+		{"Base", 8453, 50_000_000},
+		{"Base Sepolia", 84532, 10_000_000},
+		{"Unknown chain falls back to DefaultGasPrice", 99999, DefaultGasPrice},
+		{"Zero chain falls back to DefaultGasPrice", 0, DefaultGasPrice},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := GetDefaultGasPrice(tc.chainID); got != tc.want {
+				t.Errorf("GetDefaultGasPrice(%d) = %d, want %d", tc.chainID, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestLookupStablecoin(t *testing.T) {
+	cases := []struct {
+		name    string
+		chainID uint64
+		address string
+		wantOk  bool
+		wantSym string
+		wantDec uint32
+	}{
+		{"USDC mainnet", 1, "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48", true, "USDC", 6},
+		{"USDC mainnet uppercase", 1, "0xA0B86991C6218B36C1D19D4A2E9EB0CE3606EB48", true, "USDC", 6},
+		{"DAI mainnet", 1, "0x6b175474e89094c44da98b954eedeac495271d0f", true, "DAI", 18},
+		{"PYUSD mainnet", 1, "0x6c3ea9036406852006290770bedfcaba0e23a0e8", true, "PYUSD", 6},
+		{"GUSD has 2 decimals", 1, "0x056fd409e1d7a124bd7017459dfea2f387b6d5cd", true, "GUSD", 2},
+		{"USDC base", 8453, "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913", true, "USDC", 6},
+		{"USDC sepolia (Circle test)", 11155111, "0x1c7d4b196cb0c7b01d743fbc6116a902379c7238", true, "USDC", 6},
+		{"unknown ERC20 mainnet", 1, "0x0000000000000000000000000000000000000001", false, "", 0},
+		{"unknown chain", 99999, "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48", false, "", 0},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			info, ok := LookupStablecoin(tc.chainID, tc.address)
+			if ok != tc.wantOk {
+				t.Errorf("LookupStablecoin(%d, %s) ok = %v, want %v", tc.chainID, tc.address, ok, tc.wantOk)
+			}
+			if info.Symbol != tc.wantSym {
+				t.Errorf("symbol = %q, want %q", info.Symbol, tc.wantSym)
+			}
+			if info.Decimals != tc.wantDec {
+				t.Errorf("decimals = %d, want %d", info.Decimals, tc.wantDec)
+			}
+		})
+	}
+}

core/taskengine/engine.go

@@ -295,6 +295,10 @@ func New(db storage.Storage, config *config.Config, queue *apqueue.Queue, logger
 		logger: logger,
 	}
 
+	// Wire global fee rates so Summary.Fees population (in both ComposeSummary
+	// and the context-memory summarizer) uses the aggregator's configured rates.
+	SetFeeRates(config.FeeRates)
+
 	// Initialize AI summarizer (global) from aggregator config
 	// Only context-memory API is supported - all email content generation is delegated to context-memory
 	// The aggregator acts as a pass-through for the context-memory response to SendGrid
@@ -374,6 +378,10 @@ func (n *Engine) GetTenderlyClient() *TenderlyClient {
 
 func (n *Engine) SetPriceService(priceService PriceService) {
 	n.priceService = priceService
+	// Also wire as the package-level price service so Summary.Fees population
+	// (in both ComposeSummary and ContextMemorySummarizer.Summarize) can compute
+	// native-token totals from USD platform fees and value-fee legs.
+	SetPriceService(priceService)
 }
 
 func (n *Engine) Stop() {

core/taskengine/fee_estimator.go

@@ -34,6 +34,12 @@ type FeeEstimator struct {
 type PriceService interface {
 	GetNativeTokenPriceUSD(chainID int64) (*big.Float, error)
 	GetNativeTokenSymbol(chainID int64) string
+	// GetERC20PriceUSD returns the USD price for an ERC20 contract on the
+	// given chain. Implementations should short-circuit hard-coded stablecoins
+	// (LookupStablecoinSymbol) before hitting the network. Returns an error
+	// when the price isn't available — callers render "$?" rather than
+	// fabricate a number.
+	GetERC20PriceUSD(chainID int64, contractAddress string) (*big.Float, error)
 }
 
 // FeeRates holds the fee configuration for the estimator.
@@ -192,7 +198,12 @@ func (fe *FeeEstimator) EstimateFees(ctx context.Context, req *avsproto.Estimate
 	// Step 3: Value fee — workflow-level classification
 	valueFee := fe.classifyWorkflowValue(req)
 
-	nativeTokenSymbol := fe.priceService.GetNativeTokenSymbol(chainID)
+	nativeTokenSymbol := "ETH"
+	if fe.priceService != nil {
+		if sym := fe.priceService.GetNativeTokenSymbol(chainID); sym != "" {
+			nativeTokenSymbol = sym
+		}
+	}
 
 	fe.logger.Info("✅ Fee estimation completed",
 		"execution_fee_usd", fe.feeRates.ExecutionFeeUSD,

core/taskengine/fee_estimator_test.go

@@ -2,6 +2,7 @@ package taskengine
 
 import (
 	"context"
+	"fmt"
 	"math/big"
 	"testing"
 
@@ -28,6 +29,11 @@ func (mock *mockPriceService) GetNativeTokenSymbol(chainID int64) string {
 	return "ETH"
 }
 
+func (mock *mockPriceService) GetERC20PriceUSD(chainID int64, contractAddress string) (*big.Float, error) {
+	// Test mock — no ERC20 price lookups needed for fee-estimator tests.
+	return nil, fmt.Errorf("ERC20 price lookup not supported in tests")
+}
+
 func TestFeeEstimator_ChainIDDetection(t *testing.T) {
 	logger, err := sdklogging.NewZapLogger(sdklogging.Development)
 	require.NoError(t, err)