engine.go

// The core package that manage and distribute and execute task
package taskengine

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"math/big"
	"os"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/AvaProtocol/EigenLayer-AVS/core/apqueue"
	"github.com/AvaProtocol/EigenLayer-AVS/core/chainio/aa"
	"github.com/AvaProtocol/EigenLayer-AVS/core/config"
	"github.com/AvaProtocol/EigenLayer-AVS/model"
	"github.com/AvaProtocol/EigenLayer-AVS/pkg/gow"
	"github.com/AvaProtocol/EigenLayer-AVS/storage"
	sdklogging "github.com/Layr-Labs/eigensdk-go/logging"
	"github.com/allegro/bigcache/v3"
	badger "github.com/dgraph-io/badger/v4"
	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/common/math"
	"github.com/ethereum/go-ethereum/crypto"
	"github.com/ethereum/go-ethereum/ethclient"
	"github.com/oklog/ulid/v2"
	"golang.org/x/sync/singleflight"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/types/known/structpb"

	avsproto "github.com/AvaProtocol/EigenLayer-AVS/protobuf"
)

// getTaskStatusString safely converts a TaskStatus to string, handling edge cases
func getTaskStatusString(status avsproto.TaskStatus) string {
	// The crash was caused by calling .String() on an uninitialized enum
	// This function provides a safe wrapper that ensures we always get a valid string
	return status.String()
}

const (
	JobTypeExecuteTask  = "execute_task"
	DefaultLimit        = 50
	MaxSecretNameLength = 255

	EvmErc20TransferTopic0 = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
)

var (
	rpcConn *ethclient.Client
	// websocket client used for subscription
	wsEthClient  *ethclient.Client
	wsRpcURL     string
	globalLogger sdklogging.Logger

	// a global variable that we expose to our tasks. User can use `{{name}}` to access them
	// These macro are define in our aggregator yaml config file under `macros`
	macroVars    map[string]string
	macroSecrets map[string]string
	cache        *bigcache.BigCache

	// Global token enrichment service for shared token metadata and chain detection
	globalTokenService *TokenEnrichmentService

	defaultSalt = big.NewInt(0)
)

// Set a global logger for task engine
func SetLogger(mylogger sdklogging.Logger) {
	globalLogger = mylogger
}

// Set the global macro system. macros are static, immutable and available to  all tasks at runtime
func SetMacroVars(v map[string]string) {
	macroVars = v
}

func SetMacroSecrets(v map[string]string) {
	macroSecrets = v
}

// GetMacroSecret returns a secret from macros.secrets loaded into the engine.
// Returns empty string if the key is not set or macros were not initialized.
func GetMacroSecret(key string) string {
	if macroSecrets == nil || key == "" {
		return ""
	}
	return macroSecrets[key]
}

func SetCache(c *bigcache.BigCache) {
	cache = c
}

// SetTokenEnrichmentService sets the global token enrichment service used as
// the engine's default (single-chain mode, or the gateway's primary chain).
// Also adds the service to the chain-keyed registry so chain-aware callers
// can resolve it via GetTokenEnrichmentServiceForChain — this keeps the
// single-chain path covered without forcing every caller to register
// explicitly.
func SetTokenEnrichmentService(service *TokenEnrichmentService) {
	globalTokenService = service
	RegisterTokenEnrichmentService(service)
}

// GetTokenEnrichmentService returns the global token enrichment service
func GetTokenEnrichmentService() *TokenEnrichmentService {
	return globalTokenService
}

// Initialize a shared rpc client instance
func SetRpc(rpcURL string) {
	// Skip RPC initialization for test URLs to avoid external dependencies in CI
	if strings.Contains(rpcURL, "localhost") || strings.Contains(rpcURL, "127.0.0.1") || strings.Contains(rpcURL, "mock") {
		// For test environments, set rpcConn to nil and continue
		// This allows tests to run without external RPC dependencies
		rpcConn = nil
		return
	}

	// Enhanced error handling with circuit breaker pattern
	if err := rpcCallWithCircuitBreaker(func() error {
		conn, err := ethclient.Dial(rpcURL)
		if err != nil {
			return err
		}
		rpcConn = conn
		return nil
	}, "HTTP_RPC"); err != nil {
		// In CI environment, if RPC connection fails, set to nil instead of panicking
		// This allows tests to run without external dependencies
		if os.Getenv("CI") != "" || os.Getenv("GITHUB_ACTIONS") != "" {
			fmt.Printf("CI environment detected: Setting rpcConn to nil due to RPC connection failure: %v", err)
			rpcConn = nil
			return
		}

		// Log error and report to Sentry
		fmt.Printf("Failed to initialize HTTP RPC connection: %v", err)

		enhancedPanicRecovery("rpc_connection", "SetRpc", map[string]interface{}{
			"connection_type": "HTTP",
			"url":             rpcURL,
		})

		panic(fmt.Errorf("HTTP RPC connection failed: %w", err))
	}
}

// Initialize a shared websocket rpc client instance
func SetWsRpc(rpcURL string) {
	wsRpcURL = rpcURL
	// Enhanced error handling with circuit breaker pattern
	if err := rpcCallWithCircuitBreaker(retryWsRpc, "WS_RPC"); err != nil {
		// Log error instead of panic for better resilience
		fmt.Printf("Failed to initialize WebSocket RPC connection: %v", err)

		// Report to Sentry
		enhancedPanicRecovery("ws_rpc_connection", "SetWsRpc", map[string]interface{}{
			"connection_type": "WebSocket",
			"url":             rpcURL,
		})

		panic(fmt.Errorf("WebSocket RPC connection failed: %w", err))
	}
}

func retryWsRpc() error {
	for {
		conn, err := ethclient.Dial(wsRpcURL)
		if err == nil {
			wsEthClient = conn
			return nil
		}
		globalLogger.Errorf("cannot establish websocket client for RPC, retry in 15 seconds", "err", err)
		time.Sleep(15 * time.Second)
	}
}

// Minimum interval between successive connections from the same operator.
// Connections arriving faster than this are rejected with ResourceExhausted.
const operatorReconnectCooldown = 5 * time.Second

type operatorState struct {
	// list of task id that we had synced to this operator
	TaskID         map[string]bool
	MonotonicClock int64

	// Operator capabilities
	Capabilities *avsproto.SyncMessagesReq_Capabilities

	// Chains this operator advertises it can monitor.
	//
	// Empty SupportedChainIDs has TWO different meanings depending on
	// SupportedChainsExplicit:
	//   - SupportedChainsExplicit=false → legacy pre-multi-chain
	//     operator that never advertised a list. Treated as "covers
	//     everything" for back-compat.
	//   - SupportedChainsExplicit=true → the operator HAS advertised
	//     chains before (either at SyncMessages connect or via a prior
	//     Ping) and is now reporting that none of its subscriptions
	//     are live. Treated as "covers nothing" — DO NOT widen routing
	//     to this operator. The orphan-scan loop will surface the
	//     resulting tasks-with-no-coverage on the next tick.
	//
	// Once SupportedChainsExplicit is set, it stays set until the
	// operator reconnects (StreamCheckToOperator rebuilds the state
	// from scratch on a fresh stream).
	SupportedChainIDs       []int64
	SupportedChainsExplicit bool

	// Context cancellation for managing ticker lifecycle
	TickerCancel context.CancelFunc
	TickerCtx    context.Context

	// Rate limiting: track when the operator last connected
	LastConnectTime time.Time
}

type PendingNotification struct {
	TaskID    string
	Operation avsproto.MessageOp
	Timestamp time.Time
}

// The core datastructure of the task engine
type Engine struct {
	db     storage.Storage
	config *config.Config
	queue  *apqueue.Queue

	// maintain a list of active job that we have to synced to operators
	// only task triggers are sent to operator
	//
	// Lock ordering (acquire outer before inner to avoid deadlock):
	//   assignmentMutex (outer) → streamsMutex (middle) → lock (inner)
	// Never acquire an outer lock while holding an inner lock.
	// lock is a leaf lock: critical sections must only do map reads/writes,
	// never call functions that acquire streamsMutex or assignmentMutex.
	tasks            map[string]*model.Workflow
	lock             *sync.Mutex
	trackSyncedTasks map[string]*operatorState

	// orphanScanCtx / orphanScanCancel govern the periodic orphan-task
	// scan started in MustStart (gateway mode only). Cancelling lets
	// Stop() exit the loop immediately instead of waiting up to 5
	// minutes for the next tick.
	orphanScanCtx    context.Context
	orphanScanCancel context.CancelFunc

	// operator stream management for real-time notifications
	operatorStreams map[string]avsproto.Node_SyncMessagesServer
	streamsMutex    *sync.RWMutex
	// lifecycle tracking for active operator streams
	streamsWG sync.WaitGroup

	// Round-robin task assignment
	taskAssignments      map[string]string // taskID -> operatorAddress mapping
	assignmentRoundRobin int               // index for round-robin assignment
	assignmentMutex      *sync.RWMutex     // protects task assignments

	smartWalletConfig *config.SmartWalletConfig
	// chainConfigs maps chain_id to ChainConfig in gateway mode.
	// nil in single-chain mode.
	chainConfigs map[int64]*config.ChainConfig
	// when shutdown is true, our engine will perform the shutdown
	// pending execution will be pushed out before the shutdown completely
	// to force shutdown, one can type ctrl+c twice
	shutdown bool

	// seq is a monotonic number to keep track our task id
	seq storage.Sequence

	logger sdklogging.Logger

	// Token enrichment service for ERC20 transfers
	tokenEnrichmentService *TokenEnrichmentService

	// Shared clients
	tenderlyClient *TenderlyClient
	priceService   PriceService

	// Debouncing for operator approval logging
	lastApprovalLogTime map[string]time.Time
	approvalLogMutex    *sync.RWMutex

	// Deduplication for operator trigger notifications (prevents double-firing)
	// Maps trigger_request_id -> time when it was processed (for TTL cleanup)
	processedTriggerIDs map[string]time.Time
	triggerDedupLock    *sync.Mutex

	// Batched operator notifications
	pendingNotifications map[string][]PendingNotification // operatorAddr -> list of notifications
	notificationMutex    *sync.RWMutex
	notificationTicker   *time.Ticker

	// Per-execution serialization for durable resume (exactly-once). Sharded so a
	// fixed set of mutexes bounds memory — the same executionID always maps to the
	// same shard, so concurrent resumes of one execution serialize. See executionMutex.
	executionMutexes [executionLockShards]sync.Mutex

	// Collapses concurrent nodes:run requests that carry the same Idempotency-Key
	// so a retried/double-clicked Confirm can't broadcast a second UserOp. Works
	// with a persistent TTL cache (see RunNodeImmediatelyRPCIdempotent) that also
	// dedupes sequential retries after the first request has completed.
	idempotencyGroup singleflight.Group
}

// executionLockShards bounds the per-execution resume locks to a fixed set of mutexes.
const executionLockShards = 256

// executionMutex returns the lock guarding resume of executionID. Holding it across
// the load-status → run → write-terminal sequence makes durable resume exactly-once:
// a second concurrent signal (two approvers, a duplicate operator notify) blocks, then
// finds the execution already terminal and no-ops — so an on-chain ContractWrite /
// ETHTransfer in the resumed leg can never run twice.
func (n *Engine) executionMutex(executionID string) *sync.Mutex {
	// FNV-1a — allocation-free, deterministic (same id → same shard).
	var h uint32 = 2166136261
	for i := 0; i < len(executionID); i++ {
		h ^= uint32(executionID[i])
		h *= 16777619
	}
	return &n.executionMutexes[h%executionLockShards]
}

// create a new task engine using given storage, config and queue
func New(db storage.Storage, config *config.Config, queue *apqueue.Queue, logger sdklogging.Logger) *Engine {
	e := Engine{
		db:     db,
		config: config,
		queue:  queue,

		lock:                &sync.Mutex{},
		tasks:               make(map[string]*model.Workflow),
		trackSyncedTasks:    make(map[string]*operatorState),
		operatorStreams:     make(map[string]avsproto.Node_SyncMessagesServer),
		streamsMutex:        &sync.RWMutex{},
		taskAssignments:     make(map[string]string),
		assignmentMutex:     &sync.RWMutex{},
		lastApprovalLogTime: make(map[string]time.Time),
		approvalLogMutex:    &sync.RWMutex{},
		processedTriggerIDs: make(map[string]time.Time),
		triggerDedupLock:    &sync.Mutex{},
		smartWalletConfig:   config.SmartWallet,
		chainConfigs:        buildChainConfigMap(config),
		shutdown:            false,

		// Initialize batched notifications
		pendingNotifications: make(map[string][]PendingNotification),
		notificationMutex:    &sync.RWMutex{},
		notificationTicker:   time.NewTicker(3 * time.Second), // Send batched notifications every 3 seconds

		logger: logger,
	}

	// 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
	contextMemorySummarizer, err := NewContextMemorySummarizerFromAggregatorConfig(config)
	if err != nil {
		// notifications.summary is enabled but misconfigured — refuse to boot rather than
		// silently fall back, so a broken summarizer config surfaces loudly at startup.
		logger.Fatal("Invalid notifications.summary configuration — refusing to start", "error", err)
		// Defense-in-depth: Logger.Fatal is interface-defined and not guaranteed to exit on
		// every implementation (e.g. NoOpLogger.Fatal is a no-op). Force termination so a
		// misconfiguration can never fall through to the deterministic-fallback path below.
		os.Exit(1)
	}
	if contextMemorySummarizer != nil {
		SetSummarizer(contextMemorySummarizer)
		logger.Info("AI summarizer initialized", "provider", "context-memory", "base_url", config.NotificationsSummary.APIEndpoint)
	} else {
		// Summarization not enabled — the deterministic summarizer is used.
		logger.Debug("notifications.summary.enabled is false; using deterministic summarizer")
	}

	// Initialize global macro variables and secrets from config
	// This ensures all nodes (BalanceNode, etc.) can access secrets without manual setup
	SetMacroVars(config.MacroVars)
	SetMacroSecrets(config.MacroSecrets)

	SetRpc(config.SmartWallet.EthRpcUrl)
	aa.SetFactoryAddress(config.SmartWallet.FactoryAddress)
	//SetWsRpc(config.SmartWallet.EthWsUrl)

	// Use global TokenEnrichmentService or initialize if not set
	if globalTokenService == nil {
		logger.Debug("initializing global TokenEnrichmentService", "has_rpc", rpcConn != nil)
		tokenService, err := NewTokenEnrichmentService(rpcConn, logger)
		if err != nil {
			logger.Warn("Failed to initialize TokenEnrichmentService", "error", err)
			// Don't fail engine initialization, continue without token enrichment
		} else {
			globalTokenService = tokenService

			// Load token whitelist data into cache
			if err := tokenService.LoadWhitelist(); err != nil {
				logger.Warn("Failed to load token whitelist", "error", err)
				// Don't fail engine initialization, continue with RPC-only token enrichment
			}

			// Single consolidated log message
			if rpcConn != nil {
				logger.Info("Global TokenEnrichmentService initialized",
					"chainID", tokenService.GetChainID(),
					"whitelistTokens", tokenService.GetCacheSize(),
					"rpcSupport", true)
			} else {
				logger.Info("Global TokenEnrichmentService initialized",
					"chainID", tokenService.GetChainID(),
					"whitelistTokens", tokenService.GetCacheSize(),
					"rpcSupport", false)
			}
		}
	} else {
		logger.Debug("Using existing global TokenEnrichmentService")
	}
	e.tokenEnrichmentService = globalTokenService

	// Initialize shared Tenderly client from config
	e.tenderlyClient = NewTenderlyClient(config, logger)
	logger.Info("TenderlyClient initialized", "ready", e.tenderlyClient != nil)

	return &e
}

// buildChainConfigMap creates a map of chain_id -> ChainConfig from the gateway config.
// Returns nil if not in gateway mode.
func buildChainConfigMap(cfg *config.Config) map[int64]*config.ChainConfig {
	if !cfg.IsGateway || len(cfg.Chains) == 0 {
		return nil
	}
	m := make(map[int64]*config.ChainConfig, len(cfg.Chains))
	for _, chain := range cfg.Chains {
		m[chain.ChainID] = chain
	}
	return m
}

// ResolveSmartWalletConfig returns the SmartWalletConfig for a given chain_id.
// In gateway mode, it looks up the per-chain config. In single-chain mode,
// it always returns the default config.
func (n *Engine) ResolveSmartWalletConfig(chainID int64) *config.SmartWalletConfig {
	if n.chainConfigs != nil && chainID > 0 {
		if chainCfg, ok := n.chainConfigs[chainID]; ok {
			return chainCfg.SmartWallet
		}
	}
	return n.smartWalletConfig
}

// defaultChainID returns the aggregator's primary chain. Used as a fallback
// when constructing chain-scoped storage keys for entities that do not yet
// carry an explicit chain_id (back-compat for code paths that pre-date
// per-task chain ids). Returns 0 when no chain is configured — callers
// must accept 0 as a valid (placeholder) chain in that case.
func (n *Engine) defaultChainID() int64 {
	if n.smartWalletConfig != nil {
		return n.smartWalletConfig.ChainID
	}
	return 0
}

// userOwnsWalletOnAnyChain reports whether the user owns the given smart
// wallet on at least one of the chains this gateway hosts. Used by RPCs
// that query data across chains (ListWorkflowsByUser, the workflow-count
// path) — a user with tasks on multiple chains might own a wallet on only
// one of them, and we want the validation check to succeed for any.
//
// Falls through to the default-wallet equality first (the default wallet
// has the same derived address across chains when factories are aligned).
// On the first non-nil DB error, returns (false, err) — callers can
// distinguish "wallet not owned anywhere" from "DB problem."
func (n *Engine) userOwnsWalletOnAnyChain(user *model.User, walletAddr common.Address) (bool, error) {
	if user.SmartAccountAddress != nil && user.SmartAccountAddress.Hex() == walletAddr.Hex() {
		return true, nil
	}
	for _, chainID := range n.knownChainIDs() {
		ok, err := ValidWalletOwner(n.db, chainID, user, walletAddr)
		if err != nil {
			return false, err
		}
		if ok {
			return true, nil
		}
	}
	return false, nil
}

// knownChainIDs returns every chain the aggregator hosts tasks for. In
// single-chain mode that is the SmartWallet chain; in gateway mode it is
// every chain registered in chainConfigs plus the SmartWallet default.
// Used by read paths that iterate per-status prefixes — they must scan
// each chain's chain-scoped bucket separately because chain-scoped keys
// share no common prefix below "t:".
//
// When neither the SmartWallet config nor chainConfigs provides a positive
// chain_id (e.g., unit tests with a placeholder config), returns a single
// [0] entry so writes and reads use the same prefix bucket.
func (n *Engine) knownChainIDs() []int64 {
	seen := make(map[int64]struct{}, 1+len(n.chainConfigs))
	out := make([]int64, 0, 1+len(n.chainConfigs))
	if id := n.defaultChainID(); id >= 0 {
		seen[id] = struct{}{}
		out = append(out, id)
	}
	for id := range n.chainConfigs {
		if _, dup := seen[id]; dup {
			continue
		}
		seen[id] = struct{}{}
		out = append(out, id)
	}
	if len(out) == 0 {
		out = append(out, 0)
	}
	return out
}

// chainScopedTaskKey constructs the chain-scoped storage key for a task
// whose chain_id might be 0 (legacy proto / unspecified). Falls back to
// the aggregator default chain so writers never produce zero-prefixed
// (and therefore never-readable) keys.
func (n *Engine) chainScopedTaskKey(task *model.Workflow) []byte {
	return WorkflowStorageKey(task.Id, task.Status)
}

// isChainConfigured reports whether the aggregator serves chainID, using the
// canonical set knownChainIDs() (default chain + chainConfigs). ResolveSmart-
// WalletConfig is NOT a substitute — it falls back to the default config for
// an unknown chain, so it can't tell "configured" from "unconfigured".
func (n *Engine) isChainConfigured(chainID int64) bool {
	for _, id := range n.knownChainIDs() {
		if id == chainID {
			return true
		}
	}
	return false
}

// validateExplicitPartChains rejects, at create time, a task whose chain-aware
// trigger or nodes either omit chain_id (<= 0) or name a chain the aggregator
// isn't configured for. Post-G5 a task carries no chain, so every chain-aware part must name an
// explicit, configured chain; chain_id 0 or an unconfigured chain is rejected. Only enforced in gateway
// mode, where chainConfigs enumerates the served chains; single-chain mode has
// one chain and nothing to validate against.
func (n *Engine) validateExplicitPartChains(task *model.Workflow) error {
	if n.config == nil || !n.config.IsGateway || task == nil {
		return nil
	}
	check := func(kind string, chainID int64) error {
		// A task carries no chain, so every chain-aware part must name an
		// explicit, configured chain — chain_id 0 is rejected at create.
		if chainID <= 0 {
			return status.Errorf(codes.InvalidArgument,
				"%s requires an explicit chain_id (a task no longer provides a default chain)", kind)
		}
		if n.isChainConfigured(chainID) {
			return nil
		}
		return status.Errorf(codes.InvalidArgument,
			"%s targets chain_id=%d, which is not configured on this aggregator", kind, chainID)
	}
	if t := task.Trigger; t != nil {
		if et := t.GetEvent(); et != nil && et.Config != nil {
			if err := check("event trigger", et.Config.GetChainId()); err != nil {
				return err
			}
		}
		if bt := t.GetBlock(); bt != nil && bt.Config != nil {
			if err := check("block trigger", bt.Config.GetChainId()); err != nil {
				return err
			}
		}
	}
	for _, node := range task.Nodes {
		if err := checkNodeChain(node, check); err != nil {
			return err
		}
	}
	return nil
}

// checkNodeChain runs check against a node's chain-aware config, looking inside
// a Loop runner too (it wraps one chain-aware node inline).
func checkNodeChain(node *avsproto.TaskNode, check func(string, int64) error) error {
	if node == nil {
		return nil
	}
	if cw := node.GetContractWrite(); cw != nil && cw.Config != nil {
		return check("contract write node", cw.Config.GetChainId())
	}
	if cr := node.GetContractRead(); cr != nil && cr.Config != nil {
		return check("contract read node", cr.Config.GetChainId())
	}
	if et := node.GetEthTransfer(); et != nil && et.Config != nil {
		return check("eth transfer node", et.Config.GetChainId())
	}
	if await := node.GetAwait(); await != nil && await.Config != nil {
		// Only the chain-event flavor is chain-aware; external-signal Awaits have no chain.
		if ce := await.Config.GetChainEvent(); ce != nil {
			return check("await node chain event", ce.GetChainId())
		}
	}
	if loop := node.GetLoop(); loop != nil {
		if cw := loop.GetContractWrite(); cw != nil && cw.Config != nil {
			return check("loop contract write runner", cw.Config.GetChainId())
		}
		if cr := loop.GetContractRead(); cr != nil && cr.Config != nil {
			return check("loop contract read runner", cr.Config.GetChainId())
		}
		if et := loop.GetEthTransfer(); et != nil && et.Config != nil {
			return check("loop eth transfer runner", et.Config.GetChainId())
		}
	}
	// NOTE: chain-aware nodes nested inside a Branch node's conditional paths are
	// not validated here — they fall through to the strict resolveSmartWalletForNode
	// check at execution time instead of being rejected at create time. If Branch
	// gains nested chain-aware runners, add a case here (mirroring Loop above).
	return nil
}

// findTaskKey returns the chain-agnostic storage key for a task by id at the
// given status (G5: storage is no longer chain-bucketed, so the key is direct).
func (n *Engine) findTaskKey(taskID string, status avsproto.TaskStatus) []byte {
	return WorkflowStorageKey(taskID, status)
}

// taskExecutionPrefixesBytes returns the chain-agnostic execution-history
// prefix for taskID (G5: one bucket, so a single prefix).
func (n *Engine) taskExecutionPrefixesBytes(taskID string) [][]byte {
	return [][]byte{TaskExecutionPrefix(taskID)}
}

// taskExecutionPrefixes is the string form for ListKeysMulti.
func (n *Engine) taskExecutionPrefixes(taskID string) []string {
	return []string{string(TaskExecutionPrefix(taskID))}
}

// chainUserPrefixesBytes returns the chain-agnostic "u:{owner}" prefix
// (matches every task this owner has).
func (n *Engine) chainUserPrefixesBytes(owner common.Address) [][]byte {
	return [][]byte{[]byte(fmt.Sprintf("u:%s", strings.ToLower(owner.Hex())))}
}

// chainSmartWalletPrefixesBytes returns the chain-agnostic
// "u:{owner}:{wallet}" prefix (matches every task this owner has on the given
// smart wallet).
func (n *Engine) chainSmartWalletPrefixesBytes(owner common.Address, smartWallet common.Address) [][]byte {
	return [][]byte{[]byte(fmt.Sprintf("u:%s:%s",
		strings.ToLower(owner.Hex()), strings.ToLower(smartWallet.Hex())))}
}

// GetTenderlyClient returns the shared Tenderly client for fee estimation and simulation
func (n *Engine) GetTenderlyClient() *TenderlyClient {
	return n.tenderlyClient
}

func (n *Engine) SetPriceService(priceService PriceService) {
	n.priceService = priceService
}

func (n *Engine) Stop() {
	if n.seq != nil {
		if err := n.seq.Release(); err != nil {
			n.logger.Error("failed to release sequence", "error", err)
		}
	}
	// mark shutdown and collect cancels without holding locks during cancel
	var cancels []context.CancelFunc
	n.lock.Lock()
	n.shutdown = true
	for _, state := range n.trackSyncedTasks {
		if state != nil && state.TickerCancel != nil {
			cancels = append(cancels, state.TickerCancel)
		}
	}
	n.lock.Unlock()

	// cancel all operator tickers to stop their stream loops
	for _, cancel := range cancels {
		cancel()
	}

	// Stop the orphan-scan loop promptly (gateway mode only — nil
	// cancel is a no-op).
	if n.orphanScanCancel != nil {
		n.orphanScanCancel()
	}

	// wait for all StreamCheckToOperator goroutines to exit
	n.streamsWG.Wait()

	// Send any remaining notifications before shutting down
	if n.notificationTicker != nil {
		n.sendBatchedNotifications()
		n.notificationTicker.Stop()
	}
}

// chainNeedsOperatorMonitoring reports whether a trigger type needs an
// operator with a live subscription on the task's chain. Block/Event
// triggers require chain-specific operator coverage; Manual/Cron/FixedTime
// don't depend on any particular chain (cron + fixed-time still run on
// the operator's chain-agnostic TimeTrigger; manual fires entirely on
// the gateway). The question this function answers is specifically
// "does this trigger need a per-chain RPC subscription?" — not "where
// does the trigger execute?"
func chainNeedsOperatorMonitoring(tt avsproto.TriggerType) bool {
	switch tt {
	case avsproto.TriggerType_TRIGGER_TYPE_BLOCK,
		avsproto.TriggerType_TRIGGER_TYPE_EVENT:
		return true
	default:
		return false
	}
}

// triggerMonitoringChainID returns the chain an operator must subscribe to in
// order to watch this task's trigger fire (G2). For chain-watching triggers
// (event/block) it is the trigger's OWN configured chain — so a workflow can
// watch chain X while its nodes act on chain Y. Post-G5 there is no task-level
// chain: every caller passes fallbackChainID=0, and strict create-time validation
// already rejects a chain-watching trigger with chain_id<=0, so a 0 trigger chain
// here only occurs for non-chain triggers (cron/fixedtime/manual), which carry the
// fallback through unchanged — the operator's TimeTrigger is chain-agnostic and
// ignores it. Proto getters are nil-safe, so the GetEvent()/GetBlock() chain reads
// are safe for any trigger type.
func triggerMonitoringChainID(trigger *avsproto.TaskTrigger, fallbackChainID int64) int64 {
	if cid := trigger.GetEvent().GetConfig().GetChainId(); cid != 0 {
		return cid
	}
	if cid := trigger.GetBlock().GetConfig().GetChainId(); cid != 0 {
		return cid
	}
	return fallbackChainID
}

// operatorsCoveringChain returns the addresses of currently-connected
// operators that advertise the given chain_id. Empty
// SupportedChainIDs is "covers everything" ONLY when
// SupportedChainsExplicit is false — see operatorState.SupportedChainIDs
// for the explicit-empty-vs-legacy-empty distinction. Caller MUST hold
// n.lock — this reads trackSyncedTasks without locking internally.
func (n *Engine) operatorsCoveringChain(chainID int64) []string {
	if chainID == 0 {
		// Chain-agnostic tasks are covered by any connected operator;
		// returning the full set is correct.
		out := make([]string, 0, len(n.trackSyncedTasks))
		for addr := range n.trackSyncedTasks {
			out = append(out, addr)
		}
		return out
	}
	out := make([]string, 0, len(n.trackSyncedTasks))
	for addr, state := range n.trackSyncedTasks {
		if state == nil {
			continue
		}
		if len(state.SupportedChainIDs) == 0 {
			// Empty list: only treat as "covers everything" for
			// legacy operators that never explicitly advertised a
			// chain set. Explicit-empty (operator told us "I have
			// no live chains") covers nothing.
			if !state.SupportedChainsExplicit {
				out = append(out, addr)
			}
			continue
		}
		for _, id := range state.SupportedChainIDs {
			if id == chainID {
				out = append(out, addr)
				break
			}
		}
	}
	return out
}

// orphanedTaskInfo is the read-only snapshot scanOrphanedTasks
// produces while holding n.lock; logging happens after the lock is
// released so a long task list doesn't block CreateTask / Pings / the
// stream loop.
type orphanedTaskInfo struct {
	taskID      string
	chainID     int64
	triggerType string
}

// scanOrphanedTasks logs each currently-stored task whose chain is no
// longer covered by any connected operator. Runs on a slow interval
// from the engine startup loop; the alert path is the log line, not
// an automatic remediation — operators may reconnect and the orphan
// resolves itself. Logging is deliberately verbose (one line per
// orphaned task) so on-call has the task IDs at hand.
//
// The scan collects orphan info under n.lock and releases the lock
// before logging — n.logger.Warn can take non-trivial time (Sentry
// breadcrumb, sink I/O) and we don't want that blocking concurrent
// Pings / CreateTasks / stream ops.
func (n *Engine) scanOrphanedTasks() {
	orphans := n.collectOrphans()
	for _, o := range orphans {
		n.logger.Warn("🚨 Orphaned task: no connected operator advertises this chain — trigger will not fire until coverage returns",
			"task_id", o.taskID, "chain_id", o.chainID, "trigger_type", o.triggerType)
	}
}

// collectOrphans walks the task + operator state under n.lock and
// returns the list of orphaned tasks. Pure read-only — no logging or
// I/O while the lock is held.
func (n *Engine) collectOrphans() []orphanedTaskInfo {
	n.lock.Lock()
	defer n.lock.Unlock()

	// Build chain→hasOperator map up front; tasks-per-chain is much
	// larger than operators-per-chain, so this avoids re-walking the
	// operator set per task.
	chainHasOperator := make(map[int64]bool, 8)
	for _, state := range n.trackSyncedTasks {
		if state == nil {
			continue
		}
		if len(state.SupportedChainIDs) == 0 {
			// Empty list: legacy operators (never explicitly
			// advertised) still cover everything for back-compat.
			// Explicit-empty (operator told us it has no live
			// chains) covers nothing — see SupportedChainsExplicit.
			if !state.SupportedChainsExplicit {
				chainHasOperator[0] = true
			}
			continue
		}
		for _, id := range state.SupportedChainIDs {
			chainHasOperator[id] = true
		}
	}
	hasLegacyOperator := chainHasOperator[0]

	var out []orphanedTaskInfo
	for taskID, task := range n.tasks {
		if task == nil {
			continue
		}
		if task.Trigger == nil || !chainNeedsOperatorMonitoring(task.Trigger.Type) {
			continue
		}
		if hasLegacyOperator {
			continue
		}
		// Coverage is judged on the trigger's monitoring chain (G2), which is
		// where the operator actually subscribes — not the task chain.
		monitorChainID := triggerMonitoringChainID(task.Trigger, 0)
		if !chainHasOperator[monitorChainID] {
			out = append(out, orphanedTaskInfo{
				taskID:      taskID,
				chainID:     monitorChainID,
				triggerType: task.Trigger.Type.String(),
			})
		}
	}
	return out
}

// chainIDSlicesEqual compares two chain-ID slices element-wise. Order
// matters here because chainOrder is deterministic on the operator
// side — we want a reorder (which would never happen in practice) to
// still be logged as a change if it did.
func chainIDSlicesEqual(a, b []int64) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

// UpdateOperatorSupportedChains updates a connected operator's
// advertised chain capability. Called from the Ping RPC handler so a
// stalled subscription (chain configured but no recent heads — see
// operator's chainCapabilityStaleThreshold) can be dropped from
// per-chain task routing within one Ping interval, without forcing a
// SyncMessages reconnect.
//
// The operator MUST be currently connected (entry exists in
// trackSyncedTasks) for this to take effect. A Ping for an operator
// that never opened a stream is a no-op — the SyncMessages connect
// path is the authoritative one for first-time capability registration.
//
// Empty chainIDs means "operator has zero live chains right now". This
// sets SupportedChainsExplicit=true so operatorsCoveringChain treats
// the empty list as "covers nothing" rather than falling back to the
// legacy "covers everything" semantics — see operatorState
// .SupportedChainIDs.
func (n *Engine) UpdateOperatorSupportedChains(address string, chainIDs []int64) {
	n.lock.Lock()
	defer n.lock.Unlock()
	state, ok := n.trackSyncedTasks[address]
	if !ok || state == nil {
		return
	}
	// Only emit when the set actually changes — Ping runs every 5s
	// and the chain set is steady-state for the vast majority of
	// pings, so a no-op comparison shouldn't allocate.
	if !chainIDSlicesEqual(state.SupportedChainIDs, chainIDs) {
		n.logger.Info("🔁 Operator advertised chain set changed via Ping",
			"operator", address, "prev", state.SupportedChainIDs, "next", chainIDs)
	}
	state.SupportedChainIDs = chainIDs
	// Any Ping carrying the field flips the explicit flag, including
	// one with an empty list — the operator has actively told us its
	// chain capability, so we should never fall back to the legacy
	// empty-means-everything semantics for it again.
	state.SupportedChainsExplicit = true
}

// AddTaskForTesting adds a task directly to the engine's task map for testing purposes
// This bypasses database storage and validation - only use in tests
func (n *Engine) AddWorkflowForTesting(task *model.Workflow) {
	n.lock.Lock()
	defer n.lock.Unlock()
	n.tasks[task.Id] = task
}

func (n *Engine) MustStart() error {
	var err error
	n.seq, err = n.db.GetSequence([]byte("t:seq"), 1000)
	if err != nil {
		panic(err)
	}

	// Upon booting we load all enabled tasks. Storage is chain-agnostic (G5),
	// so a single "t:a:" prefix scan covers every enabled task.
	loadedCount := 0
	{
		kvs, e := n.db.GetByPrefix(WorkflowByStatusStoragePrefix(avsproto.TaskStatus_Enabled))
		if e != nil {
			panic(e)
		}
		for _, item := range kvs {
			task := &model.Workflow{
				Task: &avsproto.Task{},
			}
			// DiscardUnknown: schema evolution over time has renamed/
			// removed proto fields (e.g. `expression`, `epochs`,
			// `totalExecution`, `interval`, `input`). Strict decode
			// would silently skip any task whose body retains those
			// old fields (the `if err == nil` branch below swallows
			// errors). Tolerate unknowns at load — re-marshal on the
			// next write drops them permanently.
			err := (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(item.Value, task)
			if err == nil {
				if initErr := task.EnsureInitialized(); initErr != nil {
					n.logger.Warn("Task failed initialization after loading from storage",
						"storage_key", string(item.Key),
						"task_id", task.Id,
						"error", initErr)
					continue
				}
				n.tasks[task.Id] = task
				loadedCount++
			} else {
				n.logger.Warn("Failed to unmarshal task during startup", "storage_key", string(item.Key), "error", err)
			}
		}
	}

	n.logger.Info("🚀 Engine started successfully", "active_tasks_loaded", loadedCount)

	// Detect and handle any invalid tasks that may have been created before validation was fixed
	if err := n.DetectAndHandleInvalidTasks(); err != nil {
		n.logger.Error("Failed to handle invalid tasks during startup", "error", err)
		// Don't fail startup, but log the error
	}

	// Start the batch notification processor
	go n.processBatchedNotifications()

	// Periodic orphan scan: degradation (operator drops chain mid-run)
	// is the common failure mode, so create-time validation isn't
	// sufficient on its own. Gateway mode only — single-chain
	// deployments would log noise during operator-cold-start.
	//
	// orphanScanCtx is derived from a fresh background context (we
	// don't have a parent ctx in MustStart) and cancelled from
	// Stop() so the loop exits within milliseconds of shutdown
	// instead of waiting up to 5 minutes for the next tick.
	if n.config != nil && n.config.IsGateway {
		n.orphanScanCtx, n.orphanScanCancel = context.WithCancel(context.Background())
		go n.runOrphanScanLoop(n.orphanScanCtx)
	}

	return nil
}

// runOrphanScanLoop periodically scans persisted tasks against the
// connected-operator capability set, logging tasks whose chain has no
// coverage. 5-minute interval is generous — coverage gaps tend to
// persist (operator down) or self-heal quickly (subscription
// reconnect), so polling too aggressively just spams logs.
//
// ctx cancellation exits the loop promptly; we don't poll n.shutdown
// in between ticks because that would still leave a 5-minute
// worst-case shutdown delay.
func (n *Engine) runOrphanScanLoop(ctx context.Context) {
	ticker := time.NewTicker(5 * time.Minute)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			n.scanOrphanedTasks()
			n.sweepExpiredWaits()
		}
	}
}

// markPreviousCanonicalStaleIfAny consults the (owner, factory, salt)
// secondary index. If the index already points at a *different* wallet
// address than the freshly derived one, that means the factory's account
// implementation was upgraded between the previous record and now — the
// previously canonical wallet is no longer derivable from this triple.
// Mark the old record as stale so list responses hide it and future
// writes can replace the index entry. Index lookup or marking failures
// are logged but never block the fresh insertion path.
func (n *Engine) markPreviousCanonicalStaleIfAny(chainID int64, owner common.Address, factoryAddr common.Address, salt *big.Int, freshAddress common.Address) {
	previousAddr, err := LookupCanonicalWalletAddress(n.db, chainID, owner, factoryAddr, salt)
	if err == badger.ErrKeyNotFound {
		return
	}
	if err != nil {
		n.logger.Warn("Failed to look up canonical wallet address by (chainID, owner, factory, salt)", "chainID", chainID, "owner", owner.Hex(), "factory", factoryAddr.Hex(), "salt", salt.String(), "error", err)
		return
	}
	if strings.EqualFold(previousAddr.Hex(), freshAddress.Hex()) {
		return
	}
	n.logger.Error("Stale wallet derivation detected — factory implementation likely upgraded",
		"chainID", chainID,
		"owner", owner.Hex(),
		"factory", factoryAddr.Hex(),
		"salt", salt.String(),
		"previousAddress", previousAddr.Hex(),
		"newAddress", freshAddress.Hex(),
	)
	if markErr := MarkWalletStale(n.db, chainID, owner, previousAddr.Hex()); markErr != nil {
		n.logger.Warn("Failed to mark previous canonical wallet as stale", "chainID", chainID, "owner", owner.Hex(), "previousAddress", previousAddr.Hex(), "error", markErr)
	}
}

// storeDefaultWalletForListWallets creates and stores the default wallet (salt:0) for ListWallets.
// Returns the stored wallet model if successful, or nil if storage failed (but logs the error).
func (n *Engine) storeDefaultWalletForListWallets(chainID int64, owner common.Address, defaultDerivedAddress *common.Address, defaultSystemFactory common.Address, defaultSalt *big.Int) *model.SmartWallet {
	n.logger.Info("Default wallet not found in DB for ListWallets, storing it", "chainID", chainID, "owner", owner.Hex(), "walletAddress", defaultDerivedAddress.Hex())
	n.markPreviousCanonicalStaleIfAny(chainID, owner, defaultSystemFactory, defaultSalt, *defaultDerivedAddress)
	newModelWallet := &model.SmartWallet{
		Owner:    &owner,
		Address:  defaultDerivedAddress,
		Factory:  &defaultSystemFactory,
		Salt:     defaultSalt,
		IsHidden: false,
	}
	if storeErr := StoreWallet(n.db, chainID, owner, newModelWallet); storeErr != nil {
		n.logger.Error("Error storing default wallet to DB for ListWallets", "chainID", chainID, "owner", owner.Hex(), "walletAddress", defaultDerivedAddress.Hex(), "error", storeErr)
		// Continue and return the wallet anyway, but log the error
		return nil
	}
	// Successfully stored, return the wallet model
	return newModelWallet
}

// ListWallets corresponds to the ListWallets RPC.
//
// Wallet records are chain-scoped. The ListWalletReq proto does not carry a
// chain_id yet, so this handler queries the gateway's default chain only.
// Adding multi-chain enumeration (or a chain_id field on ListWalletReq) is
// a follow-up.
func (n *Engine) ListWallets(owner common.Address, payload *avsproto.ListWalletReq) (*avsproto.ListWalletResp, error) {
	// ListWallets takes owner directly (legacy signature) — JWT chain
	// context isn't plumbed through here yet, so fall back to the
	// gateway default. Migrating to take *model.User is a follow-up.
	chainID := n.defaultChainID()
	walletsToReturnProto := []*avsproto.SmartWallet{}
	processedAddresses := make(map[string]bool)

	defaultSystemFactory := n.smartWalletConfig.FactoryAddress
	var defaultDerivedAddress *common.Address
	var deriveErr error

	// Only try to derive default address if rpcConn is available
	if rpcConn != nil {
		defaultDerivedAddress, deriveErr = aa.GetSenderAddressForFactory(rpcConn, owner, defaultSystemFactory, defaultSalt)
	} else {
		// In test environment or when RPC is unavailable, skip default derivation
		n.logger.Debug("Skipping default wallet derivation due to nil rpcConn (test environment)", "owner", owner.Hex())
		deriveErr = fmt.Errorf("rpc connection not available")
	}

	if deriveErr != nil {
		n.logger.Warn("Failed to derive default system wallet address for ListWallets", "owner", owner.Hex(), "error", deriveErr, "hasRpcConn", rpcConn != nil)
	} else if defaultDerivedAddress == nil || *defaultDerivedAddress == (common.Address{}) {
		n.logger.Warn("Derived default system wallet address is nil or zero for ListWallets", "owner", owner.Hex())
	} else {
		includeThisDefault := true
		if payload != nil {
			if pfa := payload.GetFactoryAddress(); pfa != "" && !strings.EqualFold(defaultSystemFactory.Hex(), pfa) {
				includeThisDefault = false
			}
			if ps := payload.GetSalt(); ps != "" && defaultSalt.String() != ps {
				includeThisDefault = false
			}
		}

		if includeThisDefault {
			modelWallet, dbGetErr := GetWallet(n.db, chainID, owner, defaultDerivedAddress.Hex())

			isHidden := false
			actualSalt := defaultSalt.String()
			actualFactory := defaultSystemFactory.Hex()

			if dbGetErr == nil {
				isHidden = modelWallet.IsHidden
				actualSalt = modelWallet.Salt.String()
				if modelWallet.Factory != nil {
					actualFactory = modelWallet.Factory.Hex()
				}
			} else if dbGetErr == badger.ErrKeyNotFound {
				// Wallet not found in DB - we need to store it since we're returning it to the client
				// Enforce max smart wallet count per owner before creating a new wallet entry
				if n.smartWalletConfig != nil {
					maxAllowed := n.smartWalletConfig.MaxWalletsPerOwner
					if maxAllowed <= 0 {
						maxAllowed = config.DefaultMaxWalletsPerOwner
					}
					if maxAllowed > config.HardMaxWalletsPerOwner {
						maxAllowed = config.HardMaxWalletsPerOwner
					}
					// Fetch all wallets for this owner and count unique addresses
					dbItems, listErr := n.db.GetByPrefix(WalletByOwnerPrefix(chainID, owner))
					if listErr == nil {
						unique := make(map[string]struct{})
						for _, item := range dbItems {
							storedModelWallet := &model.SmartWallet{}
							if err := storedModelWallet.FromStorageData(item.Value); err == nil && storedModelWallet.Address != nil {
								unique[strings.ToLower(storedModelWallet.Address.Hex())] = struct{}{}
							}
						}
						// Always store the default wallet even if limit is reached, since it's always returned
						// The default wallet (salt:0) is special and should always be available
						if len(unique) >= maxAllowed {
							n.logger.Warn("Max wallet count reached for owner in ListWallets, but storing default wallet anyway since it's always returned", "owner", owner.Hex(), "limit", maxAllowed, "currentCount", len(unique))
						}
						// Store the default wallet since we're returning it to the client
						if storedWallet := n.storeDefaultWalletForListWallets(chainID, owner, defaultDerivedAddress, defaultSystemFactory, defaultSalt); storedWallet != nil {
							modelWallet = storedWallet
						}
					}
				} else {
					// No config but wallet not in DB - store it anyway
					if storedWallet := n.storeDefaultWalletForListWallets(chainID, owner, defaultDerivedAddress, defaultSystemFactory, defaultSalt); storedWallet != nil {
						modelWallet = storedWallet
					}
				}
			} else {
				n.logger.Warn("DB error fetching default derived wallet for ListWallets", "address", defaultDerivedAddress.Hex(), "error", dbGetErr)
			}

			// Always include the default wallet in the response (even if storage failed)
			if modelWallet != nil {
				isHidden = modelWallet.IsHidden
				actualSalt = modelWallet.Salt.String()
				if modelWallet.Factory != nil {
					actualFactory = modelWallet.Factory.Hex()
				}
			}

			walletsToReturnProto = append(walletsToReturnProto, &avsproto.SmartWallet{
				Address:  defaultDerivedAddress.Hex(),
				Salt:     actualSalt,
				Factory:  actualFactory,
				IsHidden: isHidden,
			})
			processedAddresses[strings.ToLower(defaultDerivedAddress.Hex())] = true
		}
	}

	dbItems, listErr := n.db.GetByPrefix(WalletByOwnerPrefix(chainID, owner))
	if listErr != nil && listErr != badger.ErrKeyNotFound {
		n.logger.Error("Error fetching wallets by owner prefix for ListWallets", "owner", owner.Hex(), "chainID", chainID, "error", listErr)
		if len(walletsToReturnProto) == 0 {
			return nil, status.Errorf(codes.Code(avsproto.ErrorCode_STORAGE_UNAVAILABLE), "Error fetching wallets by owner: %v", listErr)
		}
	}

	for _, item := range dbItems {
		storedModelWallet := &model.SmartWallet{}
		if err := storedModelWallet.FromStorageData(item.Value); err != nil {
			n.logger.Error("Failed to parse stored wallet data for ListWallets", "key", string(item.Key), "error", err)
			continue
		}

		// Hard-filter zombies left over from a factory account-implementation
		// upgrade. The record is preserved on disk (the on-chain wallet may
		// still hold assets), but it is no longer derivable from its
		// (owner, factory, salt) and would otherwise show up as a phantom
		// salt collision in the response.
		if storedModelWallet.StaleDerivation {
			continue
		}

		if processedAddresses[strings.ToLower(storedModelWallet.Address.Hex())] {
			continue
		}

		if pfa := payload.GetFactoryAddress(); pfa != "" {
			if storedModelWallet.Factory == nil || !strings.EqualFold(storedModelWallet.Factory.Hex(), pfa) {
				continue
			}
		}

		if ps := payload.GetSalt(); ps != "" {
			if storedModelWallet.Salt == nil || storedModelWallet.Salt.String() != ps {
				continue
			}
		}

		factoryString := ""
		if storedModelWallet.Factory != nil {
			factoryString = storedModelWallet.Factory.Hex()
		}
		walletsToReturnProto = append(walletsToReturnProto, &avsproto.SmartWallet{
			Address:  storedModelWallet.Address.Hex(),
			Salt:     storedModelWallet.Salt.String(),
			Factory:  factoryString,
			IsHidden: storedModelWallet.IsHidden,
		})
	}
	return &avsproto.ListWalletResp{Items: walletsToReturnProto}, nil
}

// validateNonZeroAddress validates that the factory address is not the zero address
// Returns an error if validation fails, nil if validation passes
func (n *Engine) validateNonZeroAddress(factoryAddr common.Address, methodName, ownerHex, salt string) error {
	if factoryAddr == (common.Address{}) {
		n.logger.Warn("Attempted to use zero address as factory for "+methodName, "owner", ownerHex, "salt", salt)
		return status.Errorf(codes.InvalidArgument, "Factory address cannot be the zero address")
	}
	return nil
}

// resolveUserChainID returns the chain context for a wallet RPC.
//
// REST sets user.ChainID from the JWT `aud` claim
// (aggregator/rest/middleware/jwt.go:audienceChainID); that's the
// authoritative source. Falls back to the gateway's default chain when
// user.ChainID is zero (gRPC path) or when it isn't one of the
// configured chains.
//
// The unconfigured-chain fallback matters because AuthExchange mints
// JWTs with whatever chain ID the client signed for, with no validation
// against gateway config. Without the check here, a JWT for chain
// 99999 would route storage writes into the bucket w:99999:* — orphan
// data the gateway never reads back, and unbounded keyspace growth if
// abused. Logging the fallback at Warn surfaces the anomaly so we can
// detect either a configuration drift (chain removed) or a deliberate
// probe.
func (n *Engine) resolveUserChainID(user *model.User) int64 {
	if user == nil || user.ChainID <= 0 {
		return n.defaultChainID()
	}
	for _, known := range n.knownChainIDs() {
		if known == user.ChainID {
			return user.ChainID
		}
	}
	n.logger.Warn("resolveUserChainID: user.ChainID is not a configured chain — falling back to default",
		"requested_chain_id", user.ChainID,
		"default_chain_id", n.defaultChainID(),
		"owner", user.Address.Hex(),
	)
	return n.defaultChainID()
}

// GetWallet is the gRPC handler for the GetWallet RPC.
// It uses the owner (from auth context), salt, and factory_address from payload to derive the wallet address.
//
// Wallet records are chain-scoped. The chain context is resolved via
// resolveUserChainID — REST callers route by their JWT `aud` chain,
// gRPC callers fall back to the gateway's default chain.
func (n *Engine) GetWallet(user *model.User, payload *avsproto.GetWalletReq) (*avsproto.GetWalletResp, error) {
	return n.GetWalletWithContext(context.Background(), user, payload)
}

// GetWalletWithContext is GetWallet with a caller-supplied context. REST
// handlers pass the request context so a client disconnect / timeout cancels
// the worker-routed CREATE2 derivation; the worker reader's withTimeout still
// bounds the call when ctx carries no deadline.
func (n *Engine) GetWalletWithContext(ctx context.Context, user *model.User, payload *avsproto.GetWalletReq) (*avsproto.GetWalletResp, error) {
	chainID := n.resolveUserChainID(user)
	// Resolve the per-chain smart wallet config. In single-chain mode
	// this returns n.smartWalletConfig (the gateway default); in
	// gateway/multi-chain mode it returns the config for the JWT-derived
	// chain. Factory address, RPC URL, and max-wallets-per-owner all
	// vary per chain — using the gateway default for a non-default chain
	// would derive against the wrong factory and store the resulting
	// address under the right chain key, producing inconsistent state.
	swCfg := n.ResolveSmartWalletConfig(chainID)
	if swCfg == nil {
		// Defensive — ResolveSmartWalletConfig falls back to
		// n.smartWalletConfig, which is only nil in tests with a
		// half-built Engine.
		return nil, status.Errorf(codes.Internal, "no smart wallet config for chain %d", chainID)
	}

	// Allow empty factory address (uses default), but validate non-empty ones
	if payload.GetFactoryAddress() != "" && !common.IsHexAddress(payload.GetFactoryAddress()) {
		return nil, status.Errorf(codes.InvalidArgument, InvalidFactoryAddressError)
	}

	saltBig := defaultSalt
	if payload.GetSalt() != "" {
		var ok bool
		saltBig, ok = math.ParseBig256(payload.GetSalt())
		if !ok {
			return nil, status.Errorf(codes.InvalidArgument, "%s: %s", InvalidSmartAccountSaltError, payload.GetSalt())
		}
		if saltBig.Sign() < 0 {
			return nil, status.Errorf(codes.InvalidArgument, "%s: must be a non-negative integer, got %s", InvalidSmartAccountSaltError, payload.GetSalt())
		}
	}

	factoryAddr := swCfg.FactoryAddress
	if payload.GetFactoryAddress() != "" {
		factoryAddr = common.HexToAddress(payload.GetFactoryAddress())
	}

	if err := n.validateNonZeroAddress(factoryAddr, "GetWallet", user.Address.Hex(), saltBig.String()); err != nil {
		return nil, err
	}

	var derivedSenderAddress *common.Address
	var err error

	// Wallet derivation is the on-chain source of truth — different
	// factory deployments produce different initcode and therefore
	// different CREATE2 addresses for the same (owner, salt). Persisting
	// a wrong address under the real chain key would permanently corrupt
	// the (owner, factory, salt) → wallet mapping. So the rules here are:
	//
	//  1. In test/CI environments (signalled by the package-level
	//     `rpcConn == nil` — set by initRPC when the configured RPC URL
	//     is localhost/mock or when CI dial fails), use a deterministic
	//     keccak-derived mock. This branch must NEVER fire in prod;
	//     production gateways always have a working `rpcConn`.
	//  2. In production, dial the per-chain RPC and derive against the
	//     real factory. Any failure — missing chain RPC URL, dial error,
	//     factory call error — is a HARD ERROR. Never fall back to mock.
	//
	// Using the package-level `rpcConn` for the test signal (rather than
	// `swCfg.EthRpcUrl == ""`) matches the legacy gating and avoids a
	// failure mode where a misconfigured per-chain entry silently
	// downgrades to mock derivation in production.
	if rpcConn == nil {
		n.logger.Debug("GetWallet: test/CI environment (rpcConn nil), using mock derivation",
			"owner", user.Address.Hex(), "chain_id", chainID, "salt", saltBig.String())
		concatenated := append(append(user.Address.Bytes(), factoryAddr.Bytes()...), saltBig.Bytes()...)
		hash := crypto.Keccak256(concatenated)
		mockAddr := common.BytesToAddress(hash[12:])
		derivedSenderAddress = &mockAddr
	} else if reader := GetChainStateReaderForChain(uint64(chainID)); reader != nil {
		// Preferred path: derive via the per-chain ChainStateReader. In
		// gateway mode this is worker-routed, so the gateway issues no
		// direct chain RPC for wallet derivation; in single-chain mode it's
		// a direct reader that reuses the existing client (no per-call dial).
		addr, deriveErr := reader.GetSmartWalletAddress(ctx, user.Address, factoryAddr, saltBig)
		if deriveErr != nil {
			n.logger.Error("GetWallet: factory.getAddress() failed (worker-routed)",
				"chain_id", chainID, "owner", user.Address.Hex(), "factory", factoryAddr.Hex(),
				"salt", saltBig.String(), "error", deriveErr)
			return nil, status.Errorf(codes.Unavailable,
				"GetWallet: derive sender on chain %d via factory %s: %v",
				chainID, factoryAddr.Hex(), deriveErr)
		}
		derivedSenderAddress = &addr
	} else {
		// Fallback: no ChainStateReader registered for this chain (single-
		// chain startup edge / unconfigured chain). Dial the chain RPC
		// directly, same as before the worker-routing migration.
		if swCfg.EthRpcUrl == "" {
			n.logger.Error("GetWallet: no EthRpcUrl configured for chain — refusing to derive (would risk persisting a mock address under a real chain)",
				"chain_id", chainID, "owner", user.Address.Hex())
			return nil, status.Errorf(codes.FailedPrecondition,
				"GetWallet: no EthRpcUrl configured for chain %d", chainID)
		}
		chainRPC, dialErr := ethclient.Dial(swCfg.EthRpcUrl)
		if dialErr != nil {
			n.logger.Error("GetWallet: failed to dial chain RPC",
				"chain_id", chainID, "rpc_url", swCfg.EthRpcUrl, "error", dialErr)
			return nil, status.Errorf(codes.Unavailable,
				"GetWallet: cannot reach chain %d RPC: %v", chainID, dialErr)
		}
		defer chainRPC.Close()
		derivedSenderAddress, err = aa.GetSenderAddressForFactory(chainRPC, user.Address, factoryAddr, saltBig)
		if err != nil {
			n.logger.Error("GetWallet: factory.getAddress() failed",
				"chain_id", chainID, "owner", user.Address.Hex(), "factory", factoryAddr.Hex(),
				"salt", saltBig.String(), "error", err)
			return nil, status.Errorf(codes.Unavailable,
				"GetWallet: derive sender on chain %d via factory %s: %v",
				chainID, factoryAddr.Hex(), err)
		}
	}

	if derivedSenderAddress == nil || *derivedSenderAddress == (common.Address{}) {
		n.logger.Warn("GetWallet: derived sender address is nil or zero",
			"owner", user.Address.Hex(), "factory", factoryAddr.Hex(), "salt", saltBig.String(),
			"chain_id", chainID)
		return nil, status.Errorf(codes.Internal,
			"GetWallet: derived sender address is nil or zero (chain %d, factory %s)",
			chainID, factoryAddr.Hex())
	}

	dbModelWallet, err := GetWallet(n.db, chainID, user.Address, derivedSenderAddress.Hex())

	if err != nil && err != badger.ErrKeyNotFound {
		n.logger.Error("Error fetching wallet from DB for GetWallet", "owner", user.Address.Hex(), "wallet", derivedSenderAddress.Hex(), "error", err)
		return nil, status.Errorf(codes.Code(avsproto.ErrorCode_STORAGE_UNAVAILABLE), "Error fetching wallet: %v", err)
	}

	if err == badger.ErrKeyNotFound {
		// Enforce max smart wallet count per owner before creating a new wallet entry.
		// This check only applies when creating new wallet entries, not accessing existing ones.
		// Per-chain limit comes from swCfg (resolved above) — chains can configure different
		// quotas; the gateway-default value would over-permit on quota-tightened chains.
		{
			maxAllowed := swCfg.MaxWalletsPerOwner
			if maxAllowed <= 0 {
				maxAllowed = config.DefaultMaxWalletsPerOwner
			}
			if maxAllowed > config.HardMaxWalletsPerOwner {
				maxAllowed = config.HardMaxWalletsPerOwner
			}
			// Fetch all wallets for this owner and count unique addresses
			dbItems, listErr := n.db.GetByPrefix(WalletByOwnerPrefix(chainID, user.Address))
			if listErr == nil {
				unique := make(map[string]struct{})
				for _, item := range dbItems {
					storedModelWallet := &model.SmartWallet{}
					if err := storedModelWallet.FromStorageData(item.Value); err == nil && storedModelWallet.Address != nil {
						unique[strings.ToLower(storedModelWallet.Address.Hex())] = struct{}{}
					}
				}
				if len(unique) >= maxAllowed {
					return nil, status.Errorf(codes.ResourceExhausted, "max smart wallet count reached for owner (limit=%d)", maxAllowed)
				}
			}
		}

		n.logger.Info("Wallet not found in DB for GetWallet, creating new entry", "owner", user.Address.Hex(), "walletAddress", derivedSenderAddress.Hex())
		n.markPreviousCanonicalStaleIfAny(chainID, user.Address, factoryAddr, saltBig, *derivedSenderAddress)
		newModelWallet := &model.SmartWallet{
			Owner:    &user.Address,
			Address:  derivedSenderAddress,
			Factory:  &factoryAddr,
			Salt:     saltBig,
			IsHidden: false,
		}
		if storeErr := StoreWallet(n.db, chainID, user.Address, newModelWallet); storeErr != nil {
			n.logger.Error("Error storing new wallet to DB for GetWallet", "owner", user.Address.Hex(), "walletAddress", derivedSenderAddress.Hex(), "error", storeErr)
			return nil, status.Errorf(codes.Code(avsproto.ErrorCode_STORAGE_WRITE_ERROR), "Error storing new wallet: %v", storeErr)
		}
		dbModelWallet = newModelWallet
	}

	resp := &avsproto.GetWalletResp{
		Address:        dbModelWallet.Address.Hex(),
		Salt:           dbModelWallet.Salt.String(),
		FactoryAddress: dbModelWallet.Factory.Hex(),
		IsHidden:       dbModelWallet.IsHidden,
	}

	statSvc := NewStatService(n.db)
	stat, statErr := statSvc.GetTaskCount(dbModelWallet)
	if statErr != nil {
		n.logger.Warn("Failed to get task count for GetWallet response", "walletAddress", dbModelWallet.Address.Hex(), "error", statErr)
	}
	resp.TotalTaskCount = stat.Total
	resp.EnabledTaskCount = stat.Enabled
	resp.CompletedTaskCount = stat.Completed
	resp.FailedTaskCount = stat.Failed
	resp.DisabledTaskCount = stat.Disabled

	return resp, nil
}

// GetWalletFromDB retrieves wallet information from database for validation purposes.
// Looks up the wallet on the gateway's default chain. Multi-chain wallet
// validation is a follow-up.
func (n *Engine) GetWalletFromDB(owner common.Address, smartWalletAddress string) (*model.SmartWallet, error) {
	return GetWallet(n.db, n.defaultChainID(), owner, smartWalletAddress)
}

// SetWallet is the gRPC handler for the SetWallet RPC.
// It uses the owner (from auth context), salt, and factory_address from payload to identify/derive the wallet.
// It then sets the IsHidden status for that wallet.
func (n *Engine) SetWallet(owner common.Address, payload *avsproto.SetWalletReq) (*avsproto.GetWalletResp, error) {
	// Allow empty factory address (uses default), but validate non-empty ones
	if payload.GetFactoryAddress() != "" && !common.IsHexAddress(payload.GetFactoryAddress()) {
		return nil, status.Errorf(codes.InvalidArgument, "Invalid factory address format: %s", payload.GetFactoryAddress())
	}

	saltBig, ok := math.ParseBig256(payload.GetSalt())
	if !ok {
		return nil, status.Errorf(codes.InvalidArgument, "Invalid salt format: %s", payload.GetSalt())
	}

	factoryAddr := n.smartWalletConfig.FactoryAddress // Default factory
	if payload.GetFactoryAddress() != "" {
		factoryAddr = common.HexToAddress(payload.GetFactoryAddress())
	}

	if err := n.validateNonZeroAddress(factoryAddr, "SetWallet", owner.Hex(), payload.GetSalt()); err != nil {
		return nil, err
	}

	derivedWalletAddress, err := aa.GetSenderAddressForFactory(rpcConn, owner, factoryAddr, saltBig)
	if err != nil {
		n.logger.Error("Failed to derive wallet address for SetWallet", "owner", owner.Hex(), "salt", payload.GetSalt(), "factory", payload.GetFactoryAddress(), "error", err)
		return nil, status.Errorf(codes.Internal, "Failed to derive wallet address: %v", err)
	}
	if derivedWalletAddress == nil || *derivedWalletAddress == (common.Address{}) {
		n.logger.Error("Derived wallet address is nil or zero for SetWallet", "owner", owner.Hex(), "salt", payload.GetSalt(), "factory", payload.GetFactoryAddress())
		return nil, status.Errorf(codes.Internal, "Derived wallet address is nil or zero")
	}

	err = SetWalletHiddenStatus(n.db, n.defaultChainID(), owner, derivedWalletAddress.Hex(), payload.GetIsHidden())
	if err != nil {
		if errors.Is(err, badger.ErrKeyNotFound) {
			n.logger.Warn("Wallet not found for SetWallet", "owner", owner.Hex(), "derivedAddress", derivedWalletAddress.Hex(), "salt", payload.GetSalt(), "factory", payload.GetFactoryAddress())
			// If wallet doesn't exist, SetWalletHiddenStatus from schema returns a wrapped ErrKeyNotFound.
			// The SetWallet RPC is for existing wallets identified by salt/factory.
			// So, if not found by identifiers, it's a NotFound error.
			return nil, status.Errorf(codes.NotFound, "Wallet not found for the specified salt and factory.")
		}
		n.logger.Error("Failed to set wallet hidden status via schema.SetWalletHiddenStatus", "owner", owner.Hex(), "wallet", derivedWalletAddress.Hex(), "isHidden", payload.GetIsHidden(), "error", err)
		return nil, status.Errorf(codes.Internal, "Failed to update wallet hidden status: %v", err)
	}

	updatedModelWallet, getErr := GetWallet(n.db, n.defaultChainID(), owner, derivedWalletAddress.Hex())
	if getErr != nil {
		n.logger.Error("Failed to fetch wallet after SetWallet operation", "owner", owner.Hex(), "wallet", derivedWalletAddress.Hex(), "error", getErr)
		return nil, status.Errorf(codes.Internal, "Failed to retrieve wallet details after update: %v", getErr)
	}

	resp := &avsproto.GetWalletResp{
		Address:        updatedModelWallet.Address.Hex(),
		Salt:           updatedModelWallet.Salt.String(),
		FactoryAddress: updatedModelWallet.Factory.Hex(),
		IsHidden:       updatedModelWallet.IsHidden,
	}

	statSvc := NewStatService(n.db)
	stat, statErr := statSvc.GetTaskCount(updatedModelWallet)
	if statErr != nil {
		n.logger.Warn("Failed to get task count for SetWallet response", "walletAddress", updatedModelWallet.Address.Hex(), "error", statErr)
	}
	resp.TotalTaskCount = stat.Total
	resp.EnabledTaskCount = stat.Enabled
	resp.CompletedTaskCount = stat.Completed
	resp.FailedTaskCount = stat.Failed
	resp.DisabledTaskCount = stat.Disabled

	return resp, nil
}

// resolveEventTriggerTemplates substitutes {{settings.*}} template literals in
// the EventTrigger queries' addresses and topics, mutating the trigger in place.
//
// Background: the operator subscribes to event filters at task-deploy time using
// the topics/addresses array verbatim from the stored task. It does not run the
// template resolver. If a topic slot contains an unresolved literal like
// "{{settings.runner}}", the operator cannot pad it as an address, the slot
// degenerates to a wildcard, and the filter matches any event on the contract
// instead of the intended `to=runner` constraint. This caused tasks to fire on
// unrelated transfers and exhaust their max_execution budget on noise.
//
// The simulate path (runEventTriggerImmediately in run_node_immediately.go) does
// the same resolution against a temp VM seeded with inputVariables. This helper
// is the equivalent for the persistence path used by CreateTask.
func resolveEventTriggerTemplates(trigger *avsproto.TaskTrigger, inputVariables map[string]*structpb.Value, logger sdklogging.Logger) error {
	if trigger == nil {
		return nil
	}
	eventTrigger := trigger.GetEvent()
	if eventTrigger == nil || eventTrigger.GetConfig() == nil {
		return nil
	}
	queries := eventTrigger.GetConfig().GetQueries()
	if len(queries) == 0 {
		return nil
	}

	tempVM := NewVM()
	tempVM.logger = logger
	for k, v := range inputVariables {
		tempVM.AddVar(k, v.AsInterface())
	}

	for queryIdx, q := range queries {
		for i, addr := range q.Addresses {
			if !strings.Contains(addr, "{{") {
				continue
			}
			// Resolve and always validate. If preprocessTextWithVariableMapping
			// cannot resolve the template (e.g. inputVariables is missing the
			// referenced key), it returns the original "{{...}}" string. That
			// fails IsHexAddress and triggers the error below — preventing the
			// no-op silent passthrough that would reintroduce the operator
			// wildcard bug.
			resolved := tempVM.preprocessTextWithVariableMapping(addr)
			if !common.IsHexAddress(resolved) {
				return fmt.Errorf("query[%d].addresses[%d]: template did not resolve to a valid Ethereum address: %s (original: %s)", queryIdx, i, resolved, addr)
			}
			q.Addresses[i] = resolved
			if logger != nil {
				logger.Info("EventTrigger: Resolved template in address (CreateTask)",
					"original", addr, "resolved", resolved)
			}
		}
		for i, topic := range q.Topics {
			if !strings.Contains(topic, "{{") {
				continue
			}
			// Same rationale: an unresolved "{{...}}" literal will not pass
			// validateTopicHexFormat, so a no-op resolution becomes a hard error
			// rather than a silently-broken topic filter.
			resolved := tempVM.preprocessTextWithVariableMapping(topic)
			if err := validateTopicHexFormat(resolved); err != nil {
				return fmt.Errorf("query[%d].topics[%d]: %w (original: %s, resolved: %s)", queryIdx, i, err, topic, resolved)
			}
			q.Topics[i] = resolved
			if logger != nil {
				logger.Info("EventTrigger: Resolved template in topic value (CreateTask)",
					"original", topic, "resolved", resolved)
			}
		}
	}
	return nil
}

// CreateTask records submission data
func (n *Engine) CreateWorkflow(user *model.User, taskPayload *avsproto.CreateTaskReq) (*model.Workflow, error) {
	var err error
	userAddr := user.Address.Hex()

	// Log the create task request with detailed information
	n.logger.Info("📥 CreateTask request received",
		"user", userAddr,
		"smart_wallet_address", taskPayload.SmartWalletAddress,
		"trigger_type", func() string {
			if taskPayload.Trigger != nil {
				return taskPayload.Trigger.Type.String()
			}
			return "nil"
		}(),
		"nodes_count", len(taskPayload.Nodes),
		"edges_count", len(taskPayload.Edges),
		"start_at", taskPayload.StartAt,
		"expired_at", taskPayload.ExpiredAt,
		"max_execution", taskPayload.MaxExecution,
		"name", taskPayload.Name)

	// Log node details for debugging serialization issues
	if len(taskPayload.Nodes) > 0 {
		var nodeTypes []string
		for i, node := range taskPayload.Nodes {
			nodeType := "unknown"
			if node != nil {
				nodeType = node.Type.String()
			}
			nodeTypes = append(nodeTypes, fmt.Sprintf("%d:%s", i, nodeType))
		}
		n.logger.Debug("📋 CreateTask nodes breakdown",
			"user", userAddr,
			"node_details", nodeTypes)
	}

	// Log edge details
	if len(taskPayload.Edges) > 0 {
		var edgeDetails []string
		for i, edge := range taskPayload.Edges {
			if edge != nil {
				edgeDetails = append(edgeDetails, fmt.Sprintf("%d:%s->%s", i, edge.Source, edge.Target))
			}
		}
		n.logger.Debug("🔗 CreateTask edges breakdown",
			"user", userAddr,
			"edge_details", edgeDetails)
	}

	// Validate task expiration date - must be at least 1 hour from now
	if taskPayload.ExpiredAt > 0 {
		now := time.Now().Unix() * 1000 // Convert to milliseconds for consistency with frontend
		expiredAtMs := taskPayload.ExpiredAt
		timeDifferenceMs := expiredAtMs - now
		minimumTimeMs := int64(60 * 60 * 1000) // 1 hour in milliseconds

		if timeDifferenceMs < minimumTimeMs {
			minutesRemaining := float64(timeDifferenceMs) / (60 * 1000)
			return nil, status.Errorf(codes.InvalidArgument,
				"task expiration date is too close to current time. The task must expire at least 1 hour from now. Current remaining time: %.1f minutes",
				minutesRemaining)
		}
	}

	task, err := model.NewWorkflowFromProtobuf(user, taskPayload)

	if err != nil {
		return nil, status.Errorf(codes.InvalidArgument, "%s", err.Error())
	}

	// Resolve {{settings.*}} template literals in EventTrigger query topics and
	// addresses against the task's inputVariables. The operator subscribes to
	// these filters at deploy time and cannot resolve templates itself, so an
	// unresolved literal like "{{settings.runner}}" would degrade the filter to
	// a wildcard and cause the task to fire on unrelated events. The simulate
	// path (run_node_immediately.go) does this resolution; CreateTask must too.
	if err := resolveEventTriggerTemplates(task.Trigger, task.InputVariables, n.logger); err != nil {
		return nil, status.Errorf(codes.InvalidArgument, "EventTrigger template resolution failed: %v", err)
	}

	// Default chain_id to the aggregator's SmartWallet chain when not specified.
	// In gateway mode the top-level SmartWallet is populated from chains[0]
	// (mainnet by convention), so a missing chain_id silently routes the task
	// to mainnet — which is how the user-reported Sepolia failure on
	// 2026-05-30 01:55 UTC ended up with the wrong paymaster/RPC pairing.
	// In gateway mode, require the caller to specify chain_id explicitly so
	// the task's intended chain is recorded with no ambiguity. Single-chain
	// (non-gateway) deployments keep the legacy behavior — there's only one
	// chain there, so the inference is unambiguous.
	//
	// A task no longer carries a chain (G5). Chain-aware validation below uses
	// the trigger's monitoring chain (coverage) and the aggregator default
	// chain (wallet ownership — the runner address is chain-invariant).

	// Reject tasks for chains no connected operator advertises. Block
	// and event triggers need a live operator subscription on the
	// task's chain to ever fire; without coverage the task would sit
	// in storage with `enabled` status but never execute. Only enforce
	// in gateway mode — single-chain deployments often run the
	// aggregator before the operator connects (cold-start chicken/egg)
	// and the legacy back-compat path counts no-chain-list operators
	// as covering everything.
	if n.config != nil && n.config.IsGateway && task.Trigger != nil && chainNeedsOperatorMonitoring(task.Trigger.Type) {
		// Coverage is judged on the trigger's monitoring chain (G2) — the chain
		// the operator subscribes to — not the task chain.
		monitorChainID := triggerMonitoringChainID(task.Trigger, 0)
		n.lock.Lock()
		covering := n.operatorsCoveringChain(monitorChainID)
		n.lock.Unlock()
		if len(covering) == 0 {
			n.logger.Warn("🚫 CreateTask rejected: no operator advertises this chain",
				"chain_id", monitorChainID, "trigger_type", task.Trigger.Type.String(), "user", userAddr)
			return nil, status.Errorf(codes.FailedPrecondition,
				"no connected operator currently monitors chain_id=%d for %s triggers; "+
					"task cannot fire until coverage is restored",
				monitorChainID, task.Trigger.Type.String())
		}
	}

	// Validate smart wallet ownership. The runner address comes from
	// inputVariables.settings.runner (via NewTaskFromProtobuf) and is stored
	// as task.SmartWalletAddress. We must verify the caller owns this wallet
	// on the chain this task targets.
	if task.SmartWalletAddress != "" {
		// The runner wallet may have been registered on any chain (its CREATE2
		// address is chain-invariant), and a task no longer names a chain — so
		// check ownership across every configured chain, not just the default.
		valid, ownErr := n.userOwnsWalletOnAnyChain(user, common.HexToAddress(task.SmartWalletAddress))
		if ownErr != nil {
			// Storage failure on the ownership check is NOT the same as
			// "wallet not owned" — surfacing it as InvalidArgument would
			// make a transient DB hiccup look like a malformed request.
			n.logger.Error("Wallet ownership check failed (storage error)",
				"owner", user.Address.Hex(),
				"smart_wallet", task.SmartWalletAddress,
				"chain_id", n.defaultChainID(),
				"error", ownErr)
			return nil, status.Errorf(codes.Code(avsproto.ErrorCode_STORAGE_UNAVAILABLE),
				"failed to validate smart wallet ownership: %v", ownErr)
		}
		if !valid {
			return nil, status.Errorf(codes.InvalidArgument, InvalidSmartAccountAddressError)
		}
	}

	// Validate all node names for JavaScript compatibility
	if err := validateAllNodeNamesForJavaScript(task); err != nil {
		return nil, status.Errorf(codes.InvalidArgument, "node name validation failed: %v", err)
	}

	// Reject chain-aware parts that name an explicit, unconfigured chain (G4).
	if err := n.validateExplicitPartChains(task); err != nil {
		return nil, err
	}

	// Reject an Await with an invalid config (the two flavors are mutually exclusive;
	// exactly one is required) before it can persist and fail only at execution time.
	if err := n.validateAwaitConfigs(task); err != nil {
		return nil, err
	}

	// Reject a chain-event Await whose chain no operator currently covers — a wait
	// that nothing could ever fire would never resume.
	if err := n.validateAwaitOperatorCoverage(task); err != nil {
		return nil, err
	}

	updates := map[string][]byte{}

	taskJSON, err := task.ToJSON()
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to serialize task: %v", err)
	}

	updates[string(WorkflowStorageKey(task.Id, task.Status))] = taskJSON
	updates[string(TaskUserKey(task))] = []byte(fmt.Sprintf("%d", avsproto.TaskStatus_Enabled))

	if err = n.db.BatchWrite(updates); err != nil {
		return nil, err
	}

	n.lock.Lock()
	n.tasks[task.Id] = task
	n.lock.Unlock()

	// Notify operators immediately so they can start monitoring the task
	// without waiting for the next StreamCheckToOperator ticker cycle.
	n.sendMonitorTaskTriggerToOperators(task)

	// Log successful task creation with final counts
	n.logger.Info("✅ CreateTask completed successfully",
		"user", userAddr,
		"task_id", task.Id,
		"smart_wallet_address", task.SmartWalletAddress,
		"trigger_type", task.Trigger.Type.String(),
		"final_nodes_count", len(task.Nodes),
		"final_edges_count", len(task.Edges),
		"status", getTaskStatusString(task.Status))

	return task, nil
}

func (n *Engine) StreamCheckToOperator(payload *avsproto.SyncMessagesReq, srv avsproto.Node_SyncMessagesServer) error {
	// track active stream lifecycle for graceful shutdown
	n.streamsWG.Add(1)
	defer n.streamsWG.Done()
	address := payload.Address
	connectionStartTime := time.Now()
	streamID := fmt.Sprintf("%s-%d", address[len(address)-6:], connectionStartTime.UnixNano()%10000)

	// Rate limiting: reject connections that arrive too fast from the same operator
	n.lock.Lock()
	if existing, ok := n.trackSyncedTasks[address]; ok {
		timeSinceLastConnect := connectionStartTime.Sub(existing.LastConnectTime)
		if timeSinceLastConnect < operatorReconnectCooldown {
			n.lock.Unlock()
			n.logger.Warn("⏳ Rate limiting operator reconnection - too fast",
				"operator", address,
				"stream_id", streamID,
				"time_since_last_connect", timeSinceLastConnect.String(),
				"cooldown", operatorReconnectCooldown.String())
			return status.Errorf(codes.ResourceExhausted,
				"reconnecting too fast, wait %s (last connect was %s ago)",
				operatorReconnectCooldown.String(), timeSinceLastConnect.String())
		}
		// Update timestamp under the same lock so concurrent reconnects
		// from the same operator cannot both pass the cooldown check.
		existing.LastConnectTime = connectionStartTime
	}
	n.lock.Unlock()

	n.logger.Info("open channel to stream check to operator",
		"operator", address,
		"stream_id", streamID,
		"connection_start_time", connectionStartTime.Format("15:04:05.000"),
		"monotonic_clock", payload.MonotonicClock)

	// Register this operator's stream for real-time notifications
	n.streamsMutex.Lock()
	n.operatorStreams[address] = srv
	n.streamsMutex.Unlock()

	// Create context for this connection's ticker
	tickerCtx, tickerCancel := context.WithCancel(context.Background())

	// Access and modify trackSyncedTasks under lock to avoid concurrent map access
	n.lock.Lock()
	if _, ok := n.trackSyncedTasks[address]; !ok {
		// A non-empty SupportedChainIds in the SyncMessages payload
		// means the operator has explicitly advertised its capability
		// set. Mark the flag so a subsequent Ping with an empty list
		// (all subscriptions stalled) is treated as "covers nothing"
		// rather than expanding to legacy-back-compat "covers
		// everything". An operator that never sends the field stays
		// in legacy mode.
		chains := payload.GetSupportedChainIds()
		n.trackSyncedTasks[address] = &operatorState{
			MonotonicClock:          payload.MonotonicClock,
			TaskID:                  map[string]bool{},
			Capabilities:            payload.Capabilities,
			SupportedChainIDs:       chains,
			SupportedChainsExplicit: len(chains) > 0,
			TickerCtx:               tickerCtx,
			TickerCancel:            tickerCancel,
			LastConnectTime:         connectionStartTime,
		}
		n.lock.Unlock()
		n.logger.Info("🔗 New operator connected with capabilities",
			"operator", address,
			"event_monitoring", payload.Capabilities.GetEventMonitoring(),
			"block_monitoring", payload.Capabilities.GetBlockMonitoring(),
			"time_monitoring", payload.Capabilities.GetTimeMonitoring())
	} else {
		// The operator has reconnected, cancel any existing ticker and reset state
		// Cancel old ticker if it exists (must do this while holding lock to avoid race)
		oldTickerCancel := n.trackSyncedTasks[address].TickerCancel

		if payload.MonotonicClock > n.trackSyncedTasks[address].MonotonicClock {
			n.logger.Info("🔄 Operator reconnected with newer MonotonicClock - resetting task tracking",
				"operator", address,
				"old_clock", n.trackSyncedTasks[address].MonotonicClock,
				"new_clock", payload.MonotonicClock,
				"old_task_count", len(n.trackSyncedTasks[address].TaskID))

			n.trackSyncedTasks[address].TaskID = map[string]bool{}
			n.trackSyncedTasks[address].MonotonicClock = payload.MonotonicClock
			n.trackSyncedTasks[address].Capabilities = payload.Capabilities
			reconnectChains := payload.GetSupportedChainIds()
			n.trackSyncedTasks[address].SupportedChainIDs = reconnectChains
			n.trackSyncedTasks[address].SupportedChainsExplicit = len(reconnectChains) > 0
			n.trackSyncedTasks[address].LastConnectTime = connectionStartTime

			// Set new ticker context for this connection
			n.trackSyncedTasks[address].TickerCtx = tickerCtx
			n.trackSyncedTasks[address].TickerCancel = tickerCancel

			n.logger.Info("🔄 Operator reconnected with updated capabilities",
				"operator", address,
				"event_monitoring", payload.Capabilities.GetEventMonitoring(),
				"block_monitoring", payload.Capabilities.GetBlockMonitoring(),
				"time_monitoring", payload.Capabilities.GetTimeMonitoring())
		} else {
			n.logger.Warn("⚠️ Operator reconnected with same/older MonotonicClock - force-resetting task tracking",
				"operator", address,
				"existing_clock", n.trackSyncedTasks[address].MonotonicClock,
				"new_clock", payload.MonotonicClock,
				"existing_task_count", len(n.trackSyncedTasks[address].TaskID),
				"tasks_will_be_skipped", false)

			// CRITICAL FIX: Always reset task tracking on reconnection, regardless of MonotonicClock
			// This ensures tasks are resent to reconnected operators
			n.logger.Info("🔧 Force-resetting task tracking for reconnected operator",
				"operator", address,
				"reason", "ensure_tasks_resent_on_reconnection")

			n.trackSyncedTasks[address].TaskID = map[string]bool{}
			n.trackSyncedTasks[address].Capabilities = payload.Capabilities
			forceResetChains := payload.GetSupportedChainIds()
			n.trackSyncedTasks[address].SupportedChainIDs = forceResetChains
			n.trackSyncedTasks[address].SupportedChainsExplicit = len(forceResetChains) > 0
			n.trackSyncedTasks[address].LastConnectTime = connectionStartTime

			// Set new ticker context for this connection
			n.trackSyncedTasks[address].TickerCtx = tickerCtx
			n.trackSyncedTasks[address].TickerCancel = tickerCancel
		}
		n.lock.Unlock()

		// Cancel old ticker AFTER releasing lock to avoid holding lock during cancellation
		if oldTickerCancel != nil {
			n.logger.Info("🔄 Canceling old ticker for reconnected operator",
				"operator", address,
				"old_stream", "existing")
			oldTickerCancel()
		}
	}

	// Create ticker for this connection
	ticker := time.NewTicker(15 * time.Minute)
	defer ticker.Stop()

	// Timer to trigger immediate task sending after connection stabilizes
	connectionStabilityPeriod := 10 * time.Second
	immediateSendTimer := time.NewTimer(connectionStabilityPeriod)
	defer immediateSendTimer.Stop()

	// Reset the state if the operator disconnect
	defer func() {
		n.logger.Info("🔌 Operator disconnecting, cleaning up state",
			"operator", address,
			"stream_id", streamID)

		n.lock.Lock()
		if n.trackSyncedTasks[address] != nil {
			// Only cancel the ticker context if it's OUR context (not a newer connection's context)
			if n.trackSyncedTasks[address].TickerCtx == tickerCtx {
				n.logger.Info("🔄 Canceling ticker context for this connection",
					"operator", address,
					"stream_id", streamID)
				if n.trackSyncedTasks[address].TickerCancel != nil {
					n.trackSyncedTasks[address].TickerCancel()
				}
				n.trackSyncedTasks[address].TaskID = map[string]bool{}
				n.trackSyncedTasks[address].MonotonicClock = 0
			} else {
				n.logger.Info("🔄 Skipping ticker cancellation - newer connection exists",
					"operator", address,
					"stream_id", streamID)
			}
		}
		n.lock.Unlock()

		// Only unregister the operator's stream if it's still ours
		// (a newer connection may have already replaced it)
		n.streamsMutex.Lock()
		if n.operatorStreams[address] == srv {
			delete(n.operatorStreams, address)
			n.streamsMutex.Unlock()

			// Only reassign tasks if we actually removed the stream
			n.reassignOrphanedTasks()
		} else {
			n.streamsMutex.Unlock()
			n.logger.Info("🔄 Skipping stream cleanup - newer connection already registered",
				"operator", address,
				"stream_id", streamID)
		}

		n.logger.Info("✅ Operator cleanup completed",
			"operator", address,
			"stream_id", streamID)
	}()

	//nolint:S1000
	for {
		select {
		case <-tickerCtx.Done():
			n.logger.Info("🛑 Ticker context canceled, stopping ticker loop",
				"operator", address,
				"stream_id", streamID)
			return nil
		case <-immediateSendTimer.C:
			// Immediate send triggered after connection stability period
			n.logger.Info("🚀 Immediate task send triggered for operator connection",
				"operator", address,
				"stream_id", streamID)

			n.lock.Lock()
			isShutdown := n.shutdown
			hasNoTasks := n.tasks == nil
			n.lock.Unlock()
			if isShutdown {
				return nil
			}

			if hasNoTasks {
				n.logger.Debug("📭 No tasks available",
					"operator", address)
				continue
			}
		case <-ticker.C:
			tickTime := time.Now()
			connectionAge := time.Since(connectionStartTime)

			n.logger.Info("📟 Ticker fired",
				"operator", address,
				"stream_id", streamID)

			n.lock.Lock()
			isShutdown := n.shutdown
			hasNoTasks := n.tasks == nil
			tasksCount := len(n.tasks)
			n.lock.Unlock()
			if isShutdown {
				return nil
			}

			if hasNoTasks {
				n.logger.Debug("📭 No tasks available",
					"operator", address)
				continue
			}

			// Add connection stability grace period to prevent race conditions
			if connectionAge < connectionStabilityPeriod {
				n.logger.Info("⏳ Waiting for connection to stabilize before sending tasks",
					"operator", address,
					"stream_id", streamID,
					"connection_age", connectionAge.String(),
					"min_required", "10s",
					"remaining_wait", (connectionStabilityPeriod - connectionAge).String(),
					"tick_time", tickTime.Format("15:04:05.000"))
				continue
			}

			n.logger.Info("✅ Checking task assignment",
				"operator", address,
				"tasks_count", tasksCount)
		}

		// Send tasks to operator (shared between immediate send and ticker)
		if !n.CanStreamCheck(address) {
			// This isn't a consensus approval. It's a feature flag we control server side whether to stream data to the operator or not.
			// TODO: Remove this flag when we measure performance impact on all operator

			// Use debounced logging to prevent spam (only log every 3 minutes per operator)
			if n.shouldLogApprovalMessage(address) {
				// Build dynamic approved operators list for logging
				var approvedList []string
				if len(n.config.ApprovedOperators) == 0 {
					// Use hardcoded list if no configuration
					approvedList = []string{"0x997e5d40a32c44a3d93e59fc55c4fd20b7d2d49d", "0xc6b87cc9e85b07365b6abefff061f237f7cf7dc3", "0xa026265a0f01a6e1a19b04655519429df0a57c4e"}
				} else {
					// Use configured list
					for _, addr := range n.config.ApprovedOperators {
						approvedList = append(approvedList, addr.Hex())
					}
				}

				n.logger.Info("operator has not been approved to process task",
					"operator", address,
					"approved_operators", approvedList,
					"next_log_in", "3 minutes if still not approved")
			}
			continue
		}

		// Removed excessive debug logging that was happening every 5 seconds
		// Only log when there are actual tasks to send or state changes

		// Reassign orphaned tasks when operators connect/disconnect
		// IMPORTANT: Only after connection has stabilized to prevent premature assignment
		n.reassignOrphanedTasks()

		// Aggregate tasks for this operator to reduce logging and improve efficiency
		var tasksToStream []*model.Workflow
		var tasksByTriggerType = make(map[string]int) // Count tasks by trigger type
		var newAssignments []string                   // Track new task assignments for this operator
		var orphanedTasksReclaimed []string           // Track orphaned tasks being reclaimed

		n.logger.Debug("🔍 Processing tasks for operator assignment",
			"operator", address,
			"total_tasks_in_memory", len(n.tasks))

		// Iterate over a snapshot to avoid concurrent map iteration/write panics
		n.lock.Lock()
		snapshot := make([]*model.Workflow, 0, len(n.tasks))
		for _, t := range n.tasks {
			snapshot = append(snapshot, t)
		}
		n.lock.Unlock()

		// snapshot currently tracked task IDs for this operator under lock
		n.lock.Lock()
		tracked := make(map[string]bool)
		if state, exists := n.trackSyncedTasks[address]; exists && state != nil {
			for id, present := range state.TaskID {
				if present {
					tracked[id] = true
				}
			}
		}
		n.lock.Unlock()

		for _, task := range snapshot {
			if _, ok := tracked[task.Id]; ok {
				continue
			}

			// CRITICAL FIX: Check for orphaned tasks (assigned to empty string) and reclaim them
			n.assignmentMutex.RLock()
			currentAssignment, isAssigned := n.taskAssignments[task.Id]
			n.assignmentMutex.RUnlock()

			n.logger.Debug("🔍 Checking task assignment status",
				"operator", address,
				"task_id", task.Id,
				"is_assigned", isAssigned,
				"current_assignment", currentAssignment)

			var wasReclaimed bool
			if (isAssigned && currentAssignment == "") || !isAssigned {
				// This task is orphaned (assigned to empty string) OR has no assignment at all
				// Both cases mean we should reclaim it for this reconnecting operator
				n.assignmentMutex.Lock()
				n.taskAssignments[task.Id] = address
				n.assignmentMutex.Unlock()

				orphanedTasksReclaimed = append(orphanedTasksReclaimed, task.Id)
				wasReclaimed = true

				previousAssignment := "empty_string"
				if !isAssigned {
					previousAssignment = "no_assignment"
				}

				n.logger.Info("🔄 Reclaimed orphaned task for operator",
					"task_id", task.Id,
					"operator", address,
					"previous_assignment", previousAssignment)
			}

			// Check if this operator is assigned to handle this task
			// CRITICAL FIX: Don't call assignTaskToOperator if we just reclaimed the task
			var assignedOperator string
			if wasReclaimed {
				assignedOperator = address // We just assigned it to this operator
			} else {
				assignedOperator = n.assignTaskToOperator(task)
			}

			if assignedOperator != address {
				// This task is assigned to a different operator
				n.logger.Debug("⏭️ Skipping task - assigned to different operator",
					"operator", address,
					"task_id", task.Id,
					"assigned_to", assignedOperator)
				continue
			}

			// Track this as a new assignment (unless it was a reclaimed orphan)
			isReclaimed := false
			for _, orphanedId := range orphanedTasksReclaimed {
				if orphanedId == task.Id {
					isReclaimed = true
					break
				}
			}
			if !isReclaimed {
				newAssignments = append(newAssignments, task.Id)
			}

			// Check if operator supports this trigger type
			if !n.supportsTaskTrigger(address, task) {
				n.logger.Info("⚠️ Skipping task - operator doesn't support trigger type",
					"task_id", task.Id,
					"operator", address,
					"trigger_type", task.Trigger.String())
				continue
			}

			// Skip tasks for chains this operator is not configured to monitor.
			// Pre-multi-chain operators (empty SupportedChainIDs) accept all
			// tasks for backward compatibility; tasks with chain_id=0 are
			// treated as chain-agnostic and assigned regardless.
			if !n.supportsTaskChain(address, task) {
				n.logger.Debug("⏭️ Skipping task - operator does not monitor this chain",
					"task_id", task.Id,
					"operator", address,
					"trigger_chain_id", triggerMonitoringChainID(task.Trigger, 0))
				continue
			}

			tasksToStream = append(tasksToStream, task)
			triggerTypeName := task.Trigger.String()
			tasksByTriggerType[triggerTypeName]++
		}

		// Internal triggers: synthetic single-shot event-watches standing in for
		// WAITING executions' chain-event waits (cross-chain Await). Unlike user tasks
		// (single-operator assignment), these BROADCAST to every operator covering the
		// chain — any one firing resumes the execution and the aggregator dedups the
		// rest. They flow through the same tracked-task + send path, so a boot or
		// operator reconnect re-arms every live wait automatically.
		for _, itrigger := range n.pendingChainEventTriggers() {
			if _, ok := tracked[itrigger.Id]; ok {
				continue
			}
			if !n.supportsTaskTrigger(address, itrigger) || !n.supportsTaskChain(address, itrigger) {
				continue
			}
			tasksToStream = append(tasksToStream, itrigger)
			tasksByTriggerType[itrigger.Trigger.String()]++
		}

		// Log task processing results
		n.logger.Debug("🔍 Task processing completed for operator",
			"operator", address,
			"tasks_to_stream", len(tasksToStream),
			"new_assignments", len(newAssignments),
			"orphaned_reclaimed", len(orphanedTasksReclaimed))

		// Log aggregated task assignments per operator
		if len(newAssignments) > 0 || len(orphanedTasksReclaimed) > 0 {
			n.logger.Info("🔄 Task assignments for operator",
				"operator", address,
				"stream_id", streamID,
				"operation", "MonitorTaskTrigger",
				"assigned_task_ids", newAssignments,
				"total_assignments", len(newAssignments),
				"orphaned_tasks_reclaimed", orphanedTasksReclaimed,
				"total_reclaimed", len(orphanedTasksReclaimed))
		}

		// Stream all tasks and log aggregated results
		if len(tasksToStream) > 0 {
			successCount := 0
			failedCount := 0
			var firstError error

			for _, task := range tasksToStream {
				resp := avsproto.SyncMessagesResp{
					Id: task.Id,
					Op: avsproto.MessageOp_MonitorTaskTrigger,

					TaskMetadata: &avsproto.SyncMessagesResp_TaskMetadata{
						TaskId:    task.Id,
						Remain:    task.MaxExecution,
						ExpiredAt: task.ExpiredAt,
						Trigger:   task.Trigger,
						StartAt:   task.StartAt,
						// Operators route by the trigger's monitoring chain (G2),
						// not the task chain — watch X, act Y.
						ChainId: triggerMonitoringChainID(task.Trigger, 0),
					},
				}

				// Add timeout wrapper for send operation to prevent hanging
				sendError := make(chan error, 1)
				go func() {
					sendError <- srv.Send(&resp)
				}()

				select {
				case err := <-sendError:
					if err != nil {
						failedCount++
						if firstError == nil {
							firstError = err
						}
						n.logger.Warn("⚠️ Failed to send task to operator (will retry next cycle)",
							"task_id", task.Id,
							"operator", payload.Address,
							"error", err,
							"error_type", fmt.Sprintf("%T", err),
							"grpc_status", status.Code(err).String())

						// Check if this is a connection-level error that requires reconnection
						grpcCode := status.Code(err)
						if grpcCode == codes.Unavailable || grpcCode == codes.Canceled || grpcCode == codes.DeadlineExceeded {
							n.logger.Error("🔥 Connection-level error detected, operator needs to reconnect",
								"operator", payload.Address,
								"error", err,
								"grpc_code", grpcCode.String())
							return fmt.Errorf("connection-level error, operator must reconnect: %w", err)
						}
					} else {
						n.lock.Lock()
						n.trackSyncedTasks[address].TaskID[task.Id] = true
						n.lock.Unlock()
						successCount++
					}
				case <-time.After(2 * time.Second):
					failedCount++
					n.logger.Warn("⏰ Timeout sending task to operator (will retry next cycle)",
						"task_id", task.Id,
						"operator", payload.Address,
						"timeout", "2s")
				}
			}

			// Log aggregated results instead of individual tasks
			if successCount > 0 || failedCount > 0 {
				n.logger.Info("📤 Streamed tasks to operator",
					"operator", payload.Address,
					"total_tasks", len(tasksToStream),
					"successful", successCount,
					"failed", failedCount,
					"task_breakdown", tasksByTriggerType)
			}

			// If all tasks failed with connection errors, return error to trigger reconnection
			if failedCount > 0 && successCount == 0 && firstError != nil {
				grpcCode := status.Code(firstError)
				if grpcCode == codes.Unavailable || grpcCode == codes.Canceled || grpcCode == codes.DeadlineExceeded {
					return fmt.Errorf("all task sends failed with connection error: %w", firstError)
				}
			}
		}
	}
}

// notifyOperatorsTaskOperation queues notifications for batched sending to operators
// This method is non-blocking and batches notifications for efficiency
func (n *Engine) notifyOperatorsTaskOperation(taskID string, operation avsproto.MessageOp) {
	// MonitorTaskTrigger should not use batched notifications as it requires complete task metadata
	if operation == avsproto.MessageOp_MonitorTaskTrigger {
		n.logger.Warn("❌ MonitorTaskTrigger should not be sent via batched notifications",
			"task_id", taskID,
			"operation", operation.String(),
			"solution", "MonitorTaskTrigger is handled by StreamCheckToOperator with complete metadata")
		return
	}

	// Take a snapshot of operators tracking this task under lock
	targetOperators := make([]string, 0)
	n.lock.Lock()
	for operatorAddr, operatorState := range n.trackSyncedTasks {
		if operatorState != nil {
			if _, wasTracked := operatorState.TaskID[taskID]; wasTracked {
				targetOperators = append(targetOperators, operatorAddr)
			}
		}
	}
	n.lock.Unlock()

	// Queue notifications without holding the engine lock
	n.notificationMutex.Lock()
	for _, operatorAddr := range targetOperators {
		notification := PendingNotification{
			TaskID:    taskID,
			Operation: operation,
			Timestamp: time.Now(),
		}
		n.pendingNotifications[operatorAddr] = append(n.pendingNotifications[operatorAddr], notification)
	}
	n.notificationMutex.Unlock()

	n.logger.Debug("📢 Queued notification for batching", "task_id", taskID, "operation", operation.String())
}

// processBatchedNotifications sends batched notifications to operators periodically
func (n *Engine) processBatchedNotifications() {
	defer n.notificationTicker.Stop()

	for {
		select {
		case <-n.notificationTicker.C:
			n.sendBatchedNotifications()
		case <-time.After(1 * time.Minute): // Safety check for shutdown
			n.lock.Lock()
			shuttingDown := n.shutdown
			n.lock.Unlock()
			if shuttingDown {
				n.logger.Info("🔄 Batch notification processor shutting down")
				return
			}
		}
	}
}

// sendBatchedNotifications sends accumulated notifications to operators in batches
func (n *Engine) sendBatchedNotifications() {
	n.notificationMutex.Lock()

	// Take a snapshot of pending notifications and clear the pending map
	currentBatch := make(map[string][]PendingNotification)
	for operatorAddr, notifications := range n.pendingNotifications {
		if len(notifications) > 0 {
			currentBatch[operatorAddr] = append([]PendingNotification{}, notifications...)
		}
	}
	// Clear pending notifications
	n.pendingNotifications = make(map[string][]PendingNotification)
	n.notificationMutex.Unlock()

	if len(currentBatch) == 0 {
		return // Nothing to send
	}

	// Get current operator streams
	n.streamsMutex.RLock()
	operatorStreams := make(map[string]avsproto.Node_SyncMessagesServer)
	for operatorAddr, stream := range n.operatorStreams {
		operatorStreams[operatorAddr] = stream
	}
	n.streamsMutex.RUnlock()

	// Send notifications in parallel to each operator
	var totalNotifications int
	var operatorsNotified int64
	operationCounts := make(map[string]int) // Track counts by operation type

	for operatorAddr, notifications := range currentBatch {
		totalNotifications += len(notifications)

		// Count operations by type
		for _, notification := range notifications {
			operationCounts[notification.Operation.String()]++
		}

		if stream, exists := operatorStreams[operatorAddr]; exists {
			go func(addr string, s avsproto.Node_SyncMessagesServer, notifs []PendingNotification) {
				successCount := 0
				operatorOperations := make(map[string][]string) // operation -> task_ids

				for _, notification := range notifs {
					resp := avsproto.SyncMessagesResp{
						Id: notification.TaskID,
						Op: notification.Operation,
					}

					// Only include TaskMetadata for MonitorTaskTrigger operations
					// For other operations (CancelTask, DeleteTask), TaskMetadata is not needed
					// and sending incomplete TaskMetadata causes nil pointer issues in operators
					if notification.Operation == avsproto.MessageOp_MonitorTaskTrigger {
						// For MonitorTaskTrigger, we would need complete task data
						// But batched notifications are typically for Cancel/Delete operations
						// If we ever need to batch MonitorTaskTrigger, we should fetch full task data
						n.logger.Warn("MonitorTaskTrigger should not be sent via batched notifications",
							"task_id", notification.TaskID,
							"operation", notification.Operation.String())
					}
					// Note: TaskMetadata is intentionally nil for Cancel/Delete operations

					// Use timeout for each individual notification
					done := make(chan error, 1)
					go func() {
						done <- s.Send(&resp)
					}()

					select {
					case err := <-done:
						if err != nil {
							n.logger.Debug("Failed to send batched notification",
								"operator", addr,
								"task_id", notification.TaskID,
								"operation", notification.Operation.String(),
								"error", err)
						} else {
							successCount++
							// Track successful operations by type
							operatorOperations[notification.Operation.String()] = append(
								operatorOperations[notification.Operation.String()],
								notification.TaskID)

							// Remove the task from the operator's tracking state for delete/disable operations
							if notification.Operation == avsproto.MessageOp_DisableTask ||
								notification.Operation == avsproto.MessageOp_DeleteTask {
								n.lock.Lock()
								if state, exists := n.trackSyncedTasks[addr]; exists && state != nil {
									delete(state.TaskID, notification.TaskID)
								}
								n.lock.Unlock()
							}
						}
					case <-time.After(500 * time.Millisecond):
						n.logger.Debug("Timeout sending batched notification",
							"operator", addr,
							"task_id", notification.TaskID,
							"operation", notification.Operation.String())
					}
				}

				if successCount > 0 {
					atomic.AddInt64(&operatorsNotified, 1)
					// Log operator-centric notifications with operation breakdown
					for operation, taskIDs := range operatorOperations {
						n.logger.Info("📤 Batched notifications sent to operator",
							"operator", addr,
							"operation", operation,
							"task_ids", taskIDs,
							"total_notifications", len(taskIDs))
					}
				}
			}(operatorAddr, stream, notifications)
		} else {
			n.logger.Debug("Operator stream not available for batched notifications", "operator", operatorAddr)
		}
	}

	// Log aggregated results with operation breakdown
	if totalNotifications > 0 {
		// Give goroutines time to complete, then log summary
		go func() {
			time.Sleep(1 * time.Second)
			notifiedCount := atomic.LoadInt64(&operatorsNotified)
			if notifiedCount > 0 {
				n.logger.Info("📤 Batch notification summary",
					"total_notifications", totalNotifications,
					"operators_notified", notifiedCount,
					"operation_breakdown", operationCounts)
			}
		}()
	}
}

// sendMonitorTaskTriggerToOperators sends a MonitorTaskTrigger message with full
// task metadata directly to connected operators. This is used when a task is
// newly created or re-enabled so operators begin monitoring it immediately
// instead of waiting for the next StreamCheckToOperator ticker cycle (up to 15 minutes).
// Note: operators must handle duplicate MonitorTaskTrigger messages idempotently,
// as the StreamCheckToOperator ticker may also send the same task before
// trackSyncedTasks is updated.
func (n *Engine) sendMonitorTaskTriggerToOperators(task *model.Workflow) {
	// Snapshot operator streams under read lock
	n.streamsMutex.RLock()
	operatorStreams := make(map[string]avsproto.Node_SyncMessagesServer)
	for addr, stream := range n.operatorStreams {
		operatorStreams[addr] = stream
	}
	n.streamsMutex.RUnlock()

	if len(operatorStreams) == 0 {
		n.logger.Debug("No connected operators to notify about task",
			"task_id", task.Id)
		return
	}

	resp := avsproto.SyncMessagesResp{
		Id: task.Id,
		Op: avsproto.MessageOp_MonitorTaskTrigger,
		TaskMetadata: &avsproto.SyncMessagesResp_TaskMetadata{
			TaskId:    task.Id,
			Remain:    task.MaxExecution,
			ExpiredAt: task.ExpiredAt,
			Trigger:   task.Trigger,
			StartAt:   task.StartAt,
			// Operators route by the trigger's monitoring chain (G2), not the
			// task chain — a workflow may watch chain X and act on chain Y.
			ChainId: triggerMonitoringChainID(task.Trigger, 0),
		},
	}

	successCount := 0
	for operatorAddr, stream := range operatorStreams {
		if !n.CanStreamCheck(operatorAddr) {
			continue
		}
		if !n.supportsTaskTrigger(operatorAddr, task) {
			continue
		}

		// Use timeout to prevent a stalled operator stream from blocking callers
		// (CreateTask is a user-facing gRPC call). Matches StreamCheckToOperator pattern.
		sendErr := make(chan error, 1)
		go func() { sendErr <- stream.Send(&resp) }()
		select {
		case err := <-sendErr:
			if err != nil {
				n.logger.Warn("Failed to send MonitorTaskTrigger to operator",
					"task_id", task.Id,
					"operator", operatorAddr,
					"error", err)
				continue
			}
		case <-time.After(2 * time.Second):
			n.logger.Warn("Timeout sending MonitorTaskTrigger to operator",
				"task_id", task.Id,
				"operator", operatorAddr)
			continue
		}

		successCount++
		n.lock.Lock()
		if state, exists := n.trackSyncedTasks[operatorAddr]; exists && state != nil {
			state.TaskID[task.Id] = true
		} else {
			n.logger.Warn("trackSyncedTasks entry missing after successful send; task may be re-sent on next ticker cycle",
				"task_id", task.Id,
				"operator", operatorAddr)
		}
		n.lock.Unlock()
	}

	n.logger.Info("📤 Notified operators about task",
		"task_id", task.Id,
		"operators_notified", successCount,
		"total_connected", len(operatorStreams))
}

// TODO: Merge and verify from multiple operators
func (n *Engine) AggregateChecksResult(address string, payload *avsproto.NotifyTriggersReq) error {
	_, err := n.AggregateChecksResultWithState(address, payload)
	return err
}

// AggregateChecksResultWithState processes operator trigger notifications and returns execution state info
func (n *Engine) AggregateChecksResultWithState(address string, payload *avsproto.NotifyTriggersReq) (*ExecutionState, error) {
	// Acquire lock once for all map operations to reduce lock contention
	n.lock.Lock()

	n.logger.Debug("🔄 Processing operator trigger notification",
		"operator", address,
		"task_id", payload.TaskId,
		"trigger_type", payload.TriggerType.String())

	// Update operator task tracking
	if state, exists := n.trackSyncedTasks[address]; exists {
		state.TaskID[payload.TaskId] = true
	}

	// Internal trigger (itrig_<execId>): an operator-watched chain event standing in
	// for a suspended execution's chain-event wake. Route it to resume that execution
	// rather than the normal task-execution path (it has no user workflow to enqueue).
	if executionID, ok := parseInternalTriggerTaskID(payload.TaskId); ok {
		n.lock.Unlock()
		return n.deliverChainEventWake(executionID, payload)
	}

	// Get task information to determine execution state
	task, exists := n.tasks[payload.TaskId]

	if !exists {
		// Task not found in memory - try database lookup
		n.lock.Unlock() // Release lock for database operation

		dbTask, dbErr := n.GetWorkflowByID(payload.TaskId)
		if dbErr != nil {
			// Task not found in database either - this is likely a stale operator notification
			// Log at DEBUG level to reduce noise in production
			n.logger.Debug("Operator notified about non-existent task (likely stale data)",
				"task_id", payload.TaskId,
				"operator", address,
				"memory_task_count", len(n.tasks))

			// Clean up stale task tracking for this operator
			n.lock.Lock()
			if state, exists := n.trackSyncedTasks[address]; exists {
				delete(state.TaskID, payload.TaskId)
				n.logger.Debug("Cleaned up stale task from operator tracking",
					"task_id", payload.TaskId,
					"operator", address)
			}
			n.lock.Unlock()

			// Clean up dedup entries for deleted tasks
			n.triggerDedupLock.Lock()
			for id := range n.processedTriggerIDs {
				if strings.HasPrefix(id, payload.TaskId+":") {
					delete(n.processedTriggerIDs, id)
				}
			}
			n.triggerDedupLock.Unlock()

			return &ExecutionState{
				RemainingExecutions: 0,
				TaskStillEnabled:    false,
				Status:              "not_found",
				Message:             "Task no longer exists - operator should stop monitoring",
			}, nil // Return nil error to avoid spam
		}

		// Task found in database but not in memory - add it to memory and continue
		n.lock.Lock()
		n.tasks[dbTask.Id] = dbTask
		task = dbTask
		n.lock.Unlock()

		n.logger.Info("Task recovered from database and added to memory",
			"task_id", payload.TaskId,
			"operator", address,
			"task_status", task.Status,
			"memory_task_count_after", len(n.tasks))
	} else {
		n.lock.Unlock() // Release lock after getting task
	}

	n.logger.Debug("processed aggregator check hit", "operator", address, "task_id", payload.TaskId)

	// Deduplicate trigger notifications using the trigger_request_id.
	// The operator reuses the same trigger_request_id for the same triggering event.
	// Formats: "taskID:millis" (time/cron), "taskID:blockNum" (block),
	// "taskID:txHash:logIndex" (event). The aggregator treats it as an opaque key;
	// if two notifications carry the same ID, the later one is dropped.
	if payload.TriggerRequestId != "" {
		n.triggerDedupLock.Lock()

		// Evict stale entries older than 5 minutes to prevent unbounded growth
		for id, seenAt := range n.processedTriggerIDs {
			if time.Since(seenAt) > 5*time.Minute {
				delete(n.processedTriggerIDs, id)
			}
		}

		if _, seen := n.processedTriggerIDs[payload.TriggerRequestId]; seen {
			n.triggerDedupLock.Unlock()
			n.logger.Info("Skipping duplicate trigger notification",
				"task_id", payload.TaskId,
				"operator", address,
				"trigger_request_id", payload.TriggerRequestId)

			// Return actual task state so the operator gets correct remaining/enabled info
			remaining := task.MaxExecution - task.ExecutionCount
			if remaining < 0 {
				remaining = 0
			}
			return &ExecutionState{
				RemainingExecutions: remaining,
				TaskStillEnabled:    task.IsRunable(),
				Status:              "deduplicated",
				Message:             "Duplicate trigger suppressed",
			}, nil
		}
		n.processedTriggerIDs[payload.TriggerRequestId] = time.Now()
		n.triggerDedupLock.Unlock()
	}

	// Check if task is still runnable
	if !task.IsRunable() {
		remainingExecutions := int64(0)
		status := "exhausted"
		message := "Task has reached execution limit or expired"

		if task.MaxExecution > 0 && task.ExecutionCount >= task.MaxExecution {
			status = "exhausted"
			message = fmt.Sprintf("Task has reached maximum executions (%d/%d)", task.ExecutionCount, task.MaxExecution)
		} else if task.ExpiredAt > 0 && time.Unix(task.ExpiredAt/1000, 0).Before(time.Now()) {
			status = "expired"
			message = "Task has expired"
		} else if task.StartAt > 0 && time.Now().UnixMilli() < task.StartAt {
			status = "not_started"
			message = "Task has not reached start time"
		}

		n.logger.Info("🛑 Task no longer runnable, will inform operator to stop monitoring",
			"task_id", payload.TaskId,
			"status", status,
			"execution_count", task.ExecutionCount,
			"max_execution", task.MaxExecution)

		return &ExecutionState{
			RemainingExecutions: remainingExecutions,
			TaskStillEnabled:    false,
			Status:              status,
			Message:             message,
		}, nil
	}

	// Task is still active, process the trigger
	// Create trigger data
	triggerData := &TriggerData{
		Type:   payload.TriggerType,
		Output: ExtractTriggerOutput(payload.TriggerOutput),
	}

	// Enrich EventTrigger output if TokenEnrichmentService is available and it's a Transfer event
	if payload.TriggerType == avsproto.TriggerType_TRIGGER_TYPE_EVENT {
		n.logger.Debug("processing event trigger",
			"task_id", payload.TaskId,
			"has_token_service", n.tokenEnrichmentService != nil)

		if n.tokenEnrichmentService != nil {
			if eventOutput, ok := triggerData.Output.(*avsproto.EventTrigger_Output); ok && eventOutput != nil {
				// With new structured data, we just log what we have
				hasData := eventOutput.Data != nil
				dataLength := 0
				if hasData {
					// Convert to string for logging purposes
					if dataStr, err := eventOutput.Data.MarshalJSON(); err == nil {
						dataLength = len(dataStr)
					}
				}

				n.logger.Debug("EventTrigger output with structured data",
					"task_id", payload.TaskId,
					"has_data", hasData,
					"data_length", dataLength)

				// Token enrichment is now handled during event parsing, not here
				// The structured data should already include all necessary enriched fields
			} else {
				n.logger.Debug("EventTrigger output is nil",
					"task_id", payload.TaskId)
			}
		} else {
			n.logger.Debug("TokenEnrichmentService not available for event enrichment",
				"task_id", payload.TaskId)
		}
	}

	queueTaskData := QueueExecutionData{
		TriggerType:   triggerData.Type,
		TriggerOutput: triggerData.Output,
		ExecutionID:   "",
	}
	// If there is a pre-created pending execution id for this task, consume it FIFO; otherwise create new
	// Only for operator-driven notifications path
	{
		pendingKeys, _ := n.db.GetKeyHasPrefix(PendingExecutionPrefix(task.Id))
		n.logger.Debug("🔍 Checking for pending execution IDs", "task_id", task.Id, "pending_keys_found", len(pendingKeys))
		var pickedID string
		for _, k := range pendingKeys {
			id := ExecutionIdFromPendingKey(k)
			n.logger.Debug("🔍 Found pending execution ID", "task_id", task.Id, "execution_id", id, "key", string(k))
			if pickedID == "" || id < pickedID {
				pickedID = id
			}
		}
		if pickedID != "" {
			queueTaskData.ExecutionID = pickedID
			n.logger.Debug("✅ Consuming pending execution ID", "task_id", task.Id, "execution_id", pickedID)
			// DO NOT delete pending key here - executor needs it to get pre-assigned index!
			// The pending key will be deleted by executor after using the pre-assigned index
		} else {
			queueTaskData.ExecutionID = model.GenerateID()
			n.logger.Debug("🆕 Creating new execution ID (no pending found)", "task_id", task.Id, "execution_id", queueTaskData.ExecutionID)
			// ensure pending status exists so GetExecutionStatus returns PENDING until completion
			_ = n.setExecutionStatusQueue(task, queueTaskData.ExecutionID)
		}
	}

	// For event triggers, if we have enriched data, convert it to a map format that survives JSON serialization
	if triggerData.Type == avsproto.TriggerType_TRIGGER_TYPE_EVENT {
		if eventOutput, ok := triggerData.Output.(*avsproto.EventTrigger_Output); ok {
			// Convert the enriched protobuf data to a map that will survive JSON serialization
			enrichedDataMap := buildTriggerDataMapFromProtobuf(triggerData.Type, eventOutput, n.logger)

			// Store the enriched data as a map instead of protobuf structure
			// This ensures the enriched data survives JSON serialization/deserialization
			queueTaskData.TriggerOutput = map[string]interface{}{
				"enriched_data": enrichedDataMap,
				"trigger_type":  triggerData.Type.String(),
			}

			n.logger.Debug("stored enriched event trigger data for queue execution",
				"task_id", payload.TaskId,
				"has_token_symbol", enrichedDataMap["tokenSymbol"] != nil,
				"has_value_formatted", enrichedDataMap["valueFormatted"] != nil)
		}
	}

	data, err := json.Marshal(queueTaskData)
	if err != nil {
		n.logger.Error("error serialize trigger to json", err)
		return &ExecutionState{
			RemainingExecutions: 0,
			TaskStillEnabled:    false,
			Status:              "error",
			Message:             "Failed to process trigger",
		}, err
	}

	if _, err := n.queue.Enqueue(JobTypeExecuteTask, payload.TaskId, data); err != nil {
		n.logger.Error("failed to enqueue task", "error", err, "task_id", payload.TaskId)
		return &ExecutionState{
			RemainingExecutions: 0,
			TaskStillEnabled:    false,
			Status:              "error",
			Message:             "Failed to queue execution",
		}, err
	}

	// Calculate remaining executions
	var remainingExecutions int64
	if task.MaxExecution == 0 {
		remainingExecutions = -1 // Unlimited executions
	} else {
		remainingExecutions = int64(task.MaxExecution - task.ExecutionCount - 1) // -1 because we just queued one
		if remainingExecutions < 0 {
			remainingExecutions = 0
		}
	}

	n.logger.Debug("task queued for execution",
		"task_id", payload.TaskId,
		"remaining_executions", remainingExecutions)

	return &ExecutionState{
		RemainingExecutions: remainingExecutions,
		TaskStillEnabled:    true,
		Status:              "enabled",
		Message:             "Trigger processed successfully",
	}, nil
}

func (n *Engine) ListWorkflowsByUser(user *model.User, payload *avsproto.ListTasksReq) (*avsproto.ListTasksResp, error) {
	if len(payload.SmartWalletAddress) == 0 {
		return nil, status.Errorf(codes.InvalidArgument, MissingSmartWalletAddressError)
	}

	// One prefix per (smart_wallet, chain) combo — chain-scoped user-task
	// keys (u:{chainID}:{owner}:{wallet}:{taskID}) share no common prefix
	// below "u:" so we must enumerate every chain.
	chainCount := len(n.knownChainIDs())
	prefixes := make([]string, 0, len(payload.SmartWalletAddress)*chainCount)
	for _, smartWalletAddress := range payload.SmartWalletAddress {
		if smartWalletAddress == "" {
			return nil, status.Errorf(codes.InvalidArgument, MissingSmartWalletAddressError)
		}

		if !common.IsHexAddress(smartWalletAddress) {
			return nil, status.Errorf(codes.InvalidArgument, InvalidSmartAccountAddressError)
		}

		valid, ownErr := n.userOwnsWalletOnAnyChain(user, common.HexToAddress(smartWalletAddress))
		if ownErr != nil {
			// Storage failure → surface as STORAGE_UNAVAILABLE; the
			// caller request is fine, the gateway can't currently answer.
			n.logger.Error("Wallet ownership check failed (storage error)",
				"owner", user.Address.Hex(),
				"smart_wallet", smartWalletAddress,
				"error", ownErr)
			return nil, status.Errorf(codes.Code(avsproto.ErrorCode_STORAGE_UNAVAILABLE),
				"failed to validate smart wallet ownership: %v", ownErr)
		}
		if !valid {
			return nil, status.Errorf(codes.InvalidArgument, InvalidSmartAccountAddressError)
		}

		smartWallet := common.HexToAddress(smartWalletAddress)
		for _, b := range n.chainSmartWalletPrefixesBytes(user.Address, smartWallet) {
			prefixes = append(prefixes, string(b))
		}
	}

	taskKeys, err := n.db.ListKeysMulti(prefixes)
	if err != nil {
		return nil, status.Errorf(codes.Code(avsproto.ErrorCode_STORAGE_UNAVAILABLE), StorageUnavailableError)
	}

	// second, do the sort, this is key sorted by ordering of their insertion.
	// Use the chain-aware parser so both legacy (u:{owner}:{wallet}:{taskID})
	// and chain-scoped (u:{chainID}:{owner}:{wallet}:{taskID}) keys yield
	// the same taskID for ULID comparison.
	taskIDFromKey := func(k string) string {
		parsed, perr := ParseUserTaskKey([]byte(k))
		if perr != nil && perr != ErrLegacyKey {
			return ""
		}
		return parsed.TaskID
	}
	sort.Slice(taskKeys, func(i, j int) bool {
		id1 := ulid.MustParse(taskIDFromKey(taskKeys[i]))
		id2 := ulid.MustParse(taskIDFromKey(taskKeys[j]))
		return id1.Compare(id2) < 0
	})

	taskResp := &avsproto.ListTasksResp{
		Items: []*avsproto.Task{},
		PageInfo: &avsproto.PageInfo{
			StartCursor:     "",
			EndCursor:       "",
			HasPreviousPage: false,
			HasNextPage:     false,
		},
	}

	var before, after string
	var limitVal int64

	if payload != nil {
		before = payload.Before
		after = payload.After
		limitVal = payload.Limit
	}

	cursor, limit, err := SetupPagination(before, after, limitVal)
	if err != nil {
		return nil, err
	}

	total := 0
	var hasMoreItems bool

	// Helper to load and append a task item
	appendTask := func(key string) error {
		taskID := taskIDFromKey(key)
		statusValue, err := n.db.GetKey([]byte(key))
		if err != nil {
			return err
		}
		statusInt, _ := strconv.Atoi(string(statusValue))

		taskRawByte, err := n.db.GetKey(n.findTaskKey(taskID, avsproto.TaskStatus(statusInt)))
		if err != nil {
			return nil // skip silently
		}
		task := model.NewWorkflow()
		if err := task.FromStorageData(taskRawByte); err != nil {
			return nil // skip silently
		}
		task.Id = taskID

		if t, err := task.ToProtoBuf(); err == nil {
			taskItem := &avsproto.Task{
				Id:                 t.Id,
				Owner:              t.Owner,
				SmartWalletAddress: t.SmartWalletAddress,
				StartAt:            t.StartAt,
				ExpiredAt:          t.ExpiredAt,
				Name:               t.Name,
				CompletedAt:        t.CompletedAt,
				MaxExecution:       t.MaxExecution,
				ExecutionCount:     t.ExecutionCount,
				LastRanAt:          t.LastRanAt,
				Status:             t.Status,
				Trigger:            t.Trigger,
			}

			if payload != nil {
				if payload.IncludeNodes {
					taskItem.Nodes = t.Nodes
				}
				if payload.IncludeEdges {
					taskItem.Edges = t.Edges
				}
			}

			taskResp.Items = append(taskResp.Items, taskItem)
			total += 1
		}
		return nil
	}

	if !cursor.IsZero() && cursor.Direction == CursorDirectionPrevious {
		// Backward pagination: iterate oldest-to-newest to find items
		// immediately newer than the cursor, then reverse for display order.
		cursorUlid, err := ulid.Parse(cursor.Position)
		if err != nil {
			return nil, status.Errorf(codes.InvalidArgument, InvalidCursor)
		}
		for i := 0; i < len(taskKeys); i++ {
			key := taskKeys[i]
			taskID := taskIDFromKey(key)
			taskIDUlid := model.UlidFromWorkflowId(taskID)

			// Skip items at or before the cursor position
			if taskIDUlid.Compare(cursorUlid) <= 0 {
				continue
			}

			if err := appendTask(key); err != nil {
				return nil, status.Errorf(codes.Code(avsproto.ErrorCode_STORAGE_UNAVAILABLE), StorageUnavailableError)
			}
			// Collect one extra to detect if more items exist
			if total > limit {
				break
			}
		}
		// If we collected more than limit, trim the extra and mark hasMoreItems
		if len(taskResp.Items) > limit {
			taskResp.Items = taskResp.Items[:limit]
			hasMoreItems = true
		}
		// Reverse to newest-first display order
		for i, j := 0, len(taskResp.Items)-1; i < j; i, j = i+1, j-1 {
			taskResp.Items[i], taskResp.Items[j] = taskResp.Items[j], taskResp.Items[i]
		}
	} else {
		// Forward pagination (or no cursor): iterate newest-to-oldest
		for i := len(taskKeys) - 1; i >= 0; i-- {
			key := taskKeys[i]
			taskID := taskIDFromKey(key)
			taskIDUlid := model.UlidFromWorkflowId(taskID)

			if !cursor.IsZero() {
				if cursor.LessThanOrEqualUlid(taskIDUlid) {
					continue
				}
			}

			if err := appendTask(key); err != nil {
				return nil, status.Errorf(codes.Code(avsproto.ErrorCode_STORAGE_UNAVAILABLE), StorageUnavailableError)
			}
			// Collect one extra to detect if more items exist
			if total > limit {
				break
			}
		}
		// If we collected more than limit, trim the extra and mark hasMoreItems
		if len(taskResp.Items) > limit {
			taskResp.Items = taskResp.Items[:limit]
			hasMoreItems = true
		}
	}

	// Set pagination info
	if len(taskResp.Items) > 0 {
		firstItem := taskResp.Items[0]
		lastItem := taskResp.Items[len(taskResp.Items)-1]

		taskResp.PageInfo.StartCursor = CreateNextCursor(firstItem.Id)
		taskResp.PageInfo.EndCursor = CreateNextCursor(lastItem.Id)

		if !cursor.IsZero() && cursor.Direction == CursorDirectionPrevious {
			// We reached this page via backward navigation, so the cursor position
			// and everything older always forms at least one next page.
			taskResp.PageInfo.HasNextPage = true
			taskResp.PageInfo.HasPreviousPage = hasMoreItems
		} else {
			taskResp.PageInfo.HasNextPage = hasMoreItems
			taskResp.PageInfo.HasPreviousPage = !cursor.IsZero()
		}
	}

	return taskResp, nil
}

func (n *Engine) GetWorkflowByID(taskID string) (*model.Workflow, error) {
	for statusInt := range avsproto.TaskStatus_name {
		if rawTaskData, err := n.db.GetKey(n.findTaskKey(taskID, avsproto.TaskStatus(statusInt))); err == nil {
			task := model.NewWorkflow()
			err = task.FromStorageData(rawTaskData)

			if err == nil {
				return task, nil
			}

			return nil, status.Errorf(codes.Code(avsproto.ErrorCode_TASK_DATA_CORRUPTED), TaskStorageCorruptedError)
		}
	}

	return nil, status.Errorf(codes.NotFound, TaskNotFoundError)
}

func (n *Engine) GetWorkflow(user *model.User, taskID string) (*model.Workflow, error) {
	task, err := n.GetWorkflowByID(taskID)
	if err != nil {
		return nil, err
	}

	// Admin users (authenticated with API key) use zero address and can access any task
	isAdminUser := user.Address == (common.Address{})

	if !isAdminUser && !task.OwnedBy(user.Address) {
		return nil, status.Errorf(codes.NotFound, TaskNotFoundError)
	}

	return task, nil
}

// instructOperatorImmediateTrigger instructs the connected operator to trigger the task immediately
// with the current block number, bypassing the normal interval waiting
func (n *Engine) instructOperatorImmediateTrigger(ctx context.Context, taskID string) error {
	// Load the task first so the current-block read targets the task's
	// chain (via its per-chain worker), not a single gateway RPC.
	task, err := n.GetWorkflowByID(taskID)
	if err != nil {
		return fmt.Errorf("failed to get task for immediate trigger: %w", err)
	}

	// The current-block read must target the block trigger's own chain (G5) —
	// a zero/unset chain is a hard error so the read never silently targets the
	// wrong chain.
	chainID := triggerMonitoringChainID(task.Trigger, 0)
	if chainID == 0 {
		return fmt.Errorf("task %s block trigger has no chain_id; cannot resolve chain for immediate block trigger", taskID)
	}
	reader := GetChainStateReaderForChain(uint64(chainID))
	if reader == nil {
		return fmt.Errorf("no chain-state reader available for chain %d (immediate block trigger)", chainID)
	}
	currentBlock, err := reader.GetBlockNumber(ctx)
	if err != nil {
		return fmt.Errorf("failed to get current block number: %w", err)
	}

	if n.logger != nil {
		n.logger.Info("Instructing operator to trigger immediately",
			"task_id", taskID,
			"chain_id", chainID,
			"current_block", currentBlock)
	}

	// Send immediate trigger instruction directly to operators with task metadata
	// This bypasses the batched notification system to ensure operators get complete task information
	n.streamsMutex.RLock()
	operatorStreams := make(map[string]avsproto.Node_SyncMessagesServer)
	for operatorAddr, stream := range n.operatorStreams {
		operatorStreams[operatorAddr] = stream
	}
	n.streamsMutex.RUnlock()

	// Send immediate trigger notification to ALL connected operators
	// For immediate triggers, we don't wait for task assignment - we send to all connected operators
	// The operators will handle the trigger if they're capable, regardless of current task assignments
	n.lock.Lock()
	allConnectedOperators := make([]string, 0)
	for operatorAddr := range n.trackSyncedTasks {
		allConnectedOperators = append(allConnectedOperators, operatorAddr)
	}
	n.lock.Unlock()

	// Send immediate trigger notification with complete task metadata to all connected operators
	successCount := 0
	for _, operatorAddr := range allConnectedOperators {
		if stream, exists := operatorStreams[operatorAddr]; exists {
			resp := avsproto.SyncMessagesResp{
				Id: taskID,
				Op: avsproto.MessageOp_ImmediateTrigger,
				TaskMetadata: &avsproto.SyncMessagesResp_TaskMetadata{
					TaskId:    task.Id,
					Remain:    task.MaxExecution,
					ExpiredAt: task.ExpiredAt,
					Trigger:   task.Trigger,
					StartAt:   task.StartAt,
					// Operators route by the trigger's monitoring chain (G2).
					ChainId: triggerMonitoringChainID(task.Trigger, 0),
				},
			}

			// Send with timeout to prevent hanging
			done := make(chan error, 1)
			go func() {
				done <- stream.Send(&resp)
			}()

			select {
			case err := <-done:
				if err != nil {
					if n.logger != nil {
						n.logger.Error("Failed to send immediate trigger notification to operator",
							"operator", operatorAddr,
							"task_id", taskID,
							"error", err)
					}
				} else {
					successCount++
					if n.logger != nil {
						n.logger.Debug("✅ Sent immediate trigger notification to operator",
							"operator", operatorAddr,
							"task_id", taskID,
							"current_block", currentBlock)
					}
				}
			case <-time.After(2 * time.Second):
				if n.logger != nil {
					n.logger.Warn("⏰ Timeout sending immediate trigger notification to operator",
						"operator", operatorAddr,
						"task_id", taskID,
						"timeout", "2s")
				}
			}
		}
	}

	if successCount == 0 {
		return fmt.Errorf("failed to send immediate trigger notification to any operator")
	}

	if n.logger != nil {
		n.logger.Info("✅ Successfully sent immediate trigger notifications",
			"task_id", taskID,
			"operators_notified", successCount,
			"total_connected_operators", len(allConnectedOperators))
	}

	return nil
}

func (n *Engine) TriggerWorkflow(user *model.User, payload *avsproto.TriggerTaskReq) (*avsproto.TriggerTaskResp, error) {
	return n.TriggerWorkflowWithContext(context.Background(), user, payload)
}

// TriggerWorkflowWithContext is TriggerWorkflow with a caller-supplied context.
// REST handlers pass the request context so a client disconnect / timeout
// cancels the worker-routed current-block read (immediate trigger) and the
// blocking execution's wallet-ownership validation.
func (n *Engine) TriggerWorkflowWithContext(ctx context.Context, user *model.User, payload *avsproto.TriggerTaskReq) (*avsproto.TriggerTaskResp, error) {
	// Validate task ID format first
	if !ValidateTaskId(payload.TaskId) {
		return nil, status.Errorf(codes.InvalidArgument, InvalidTaskIdFormat)
	}

	task, err := n.GetWorkflow(user, payload.TaskId)
	if err != nil {
		return nil, err
	}

	// Explicit ownership validation for security (even though GetTask already checks this)
	if !task.OwnedBy(user.Address) {
		return nil, status.Errorf(codes.NotFound, TaskNotFoundError)
	}

	// A task no longer carries a chain (G5), so there is nothing to override —
	// the payload's chain_id is ignored. Each chain-aware node runs on its own
	// configured chain.

	// Important business logic validation: Check if task is runnable
	if !task.IsRunable() {
		return nil, status.Errorf(codes.FailedPrecondition, TaskIsNotRunnable)
	}

	// For manual block triggers, instruct the operator to trigger immediately with current block
	if payload.TriggerType == avsproto.TriggerType_TRIGGER_TYPE_BLOCK {
		if n.logger != nil {
			n.logger.Debug("Manual block trigger detected - instructing operator to trigger immediately",
				"task_id", task.Id,
				"user", user.Address.String())
		}

		// For non-blocking mode, pre-create execution ID, persist Pending status and queue marker,
		// then instruct operator and return immediately (do NOT enqueue here).
		if !payload.IsBlocking {
			preExecID := model.GenerateID()
			n.logger.Debug("🔄 Pre-creating execution ID for non-blocking trigger", "task_id", task.Id, "execution_id", preExecID)

			// Pre-assign atomic execution index for stable indexing
			preAssignedIndex, indexErr := n.AssignNextExecutionIndex(task)
			if indexErr != nil {
				n.logger.Error("Failed to assign execution index for non-blocking trigger", "task_id", task.Id, "execution_id", preExecID, "error", indexErr)
				return nil, status.Errorf(codes.Internal, "failed to assign execution index: %v", indexErr)
			}
			n.logger.Debug("🔢 Pre-assigned execution index", "task_id", task.Id, "execution_id", preExecID, "index", preAssignedIndex)

			// Mark pending in queue so GetExecutionStatus returns PENDING
			if err := n.setExecutionStatusQueue(task, preExecID); err != nil {
				return nil, err
			}
			// Persist pending execution ID with its assigned index for FIFO consumption when operator notifies
			pendingKey := PendingExecutionKey(task, preExecID)
			pendingData := fmt.Sprintf("%d", preAssignedIndex) // Store the index as the value
			if err := n.db.Set(pendingKey, []byte(pendingData)); err != nil {
				n.logger.Error("Failed to persist pending execution id with index", "task_id", task.Id, "execution_id", preExecID, "index", preAssignedIndex, "error", err)
				return nil, status.Errorf(codes.Internal, "failed to persist pending execution id")
			}
			n.logger.Debug("💾 Persisted pending execution ID with index", "task_id", task.Id, "execution_id", preExecID, "index", preAssignedIndex, "key", string(pendingKey))

			// Best-effort operator instruction (ignore errors; operator may connect later)
			_ = n.instructOperatorImmediateTrigger(ctx, task.Id)

			startTime := time.Now().UnixMilli()
			resp := &avsproto.TriggerTaskResp{
				ExecutionId: preExecID,
				WorkflowId:  payload.TaskId,
				Status:      avsproto.ExecutionStatus_EXECUTION_STATUS_PENDING,
				StartAt:     &startTime,
			}
			return resp, nil
		}

		// For blocking mode, continue with normal execution flow even for block triggers
	}

	// For non-block triggers, use the original logic
	triggerData := &TriggerData{
		Type:   payload.TriggerType,
		Output: ExtractTriggerOutput(payload.TriggerOutput),
	}

	// Validate manual trigger data if present
	if triggerData.Type == avsproto.TriggerType_TRIGGER_TYPE_MANUAL {
		if manualOutput, ok := triggerData.Output.(*avsproto.ManualTrigger_Output); ok {
			if err := ValidateManualTriggerFromProtobuf(manualOutput, task); err != nil {
				return nil, err
			}
		}
	}

	// RPC input map from TriggerTaskReq: proto/json field remains trigger_input (JSON key triggerInput)
	// for backward compatibility; semantically same role as input_variables on SimulateTaskReq.
	var inputVariables map[string]interface{}
	if len(payload.TriggerInput) > 0 {
		inputVariables = make(map[string]interface{})
		for key, value := range payload.TriggerInput {
			inputVariables[key] = value.AsInterface()
		}
	}

	queueTaskData := QueueExecutionData{
		TriggerType:    triggerData.Type,
		TriggerOutput:  triggerData.Output,
		ExecutionID:    model.GenerateID(),
		InputVariables: inputVariables,
	}

	// Store execution status as pending first
	err = n.setExecutionStatusQueue(task, queueTaskData.ExecutionID)
	if err != nil {
		return nil, err
	}

	// Create base response with always-available fields
	response := &avsproto.TriggerTaskResp{
		ExecutionId: queueTaskData.ExecutionID,
		WorkflowId:  payload.TaskId, // taskId and workflowId refer to the same entity. Consider renaming for consistency.
	}

	if payload.IsBlocking {
		executor := NewExecutor(n.ResolveSmartWalletConfig(n.defaultChainID()), n.db, n.logger, n, n.priceService)
		execution, runErr := executor.RunTaskWithContext(ctx, task, &queueTaskData)
		if runErr != nil {
			n.logger.Error("failed to run blocking task", runErr)
			// For blocking execution, return the error to the caller
			return nil, runErr
		}

		// Clean up TaskTriggerKey after successful blocking execution
		if queueTaskData.ExecutionID != "" {
			triggerKeyToClean := TaskTriggerKey(task, queueTaskData.ExecutionID)
			if delErr := n.db.Delete(triggerKeyToClean); delErr != nil {
				n.logger.Error("TriggerTask: Failed to delete TaskTriggerKey after successful blocking execution",
					"key", string(triggerKeyToClean), "task_id", task.Id, "execution_id", queueTaskData.ExecutionID, "error", delErr)
			}
		}

		if execution != nil {
			// For blocking mode, populate all execution fields like getExecution response
			response.Status = execution.Status
			response.StartAt = &execution.StartAt
			response.EndAt = &execution.EndAt
			if execution.Error != "" {
				response.Error = &execution.Error
			}
			response.Steps = execution.Steps

			// EXTRA SAFETY: persist execution synchronously for immediate ListExecutions visibility
			// Sanitize before persistence as protobuf Value rejects NaN/Inf
			sanitizeExecutionForPersistence(execution)
			mo := protojson.MarshalOptions{UseProtoNames: true, EmitUnpopulated: true}
			if b, mErr := mo.Marshal(execution); mErr == nil {
				key := TaskExecutionKey(task, execution.Id)
				if setErr := n.db.Set(key, b); setErr != nil {
					n.logger.Error("TriggerTask: failed to persist execution synchronously", "task_id", task.Id, "execution_id", execution.Id, "error", setErr)
				} else if n.logger != nil {
					n.logger.Info("TriggerTask: persisted execution synchronously", "task_id", task.Id, "execution_id", execution.Id, "key", string(key))
				}
			} else if n.logger != nil {
				n.logger.Error("TriggerTask: failed to marshal execution for persistence", "task_id", task.Id, "execution_id", execution.Id, "error", mErr)
			}
			return response, nil
		}
	} else {
		// For non-blocking mode, only set startAt (execution has started)
		startTime := time.Now().UnixMilli()
		response.StartAt = &startTime
	}

	// Add async execution
	// Ensure the execution is also persisted synchronously when blocking is requested
	// (Already handled above). For non-blocking, enqueue the job as usual.
	data, err := json.Marshal(queueTaskData)
	if err != nil {
		n.logger.Error("error serialize trigger to json", "error", err)
		return nil, err
	}

	if _, err := n.queue.Enqueue(JobTypeExecuteTask, payload.TaskId, data); err != nil {
		n.logger.Error("failed to enqueue task", "error", err, "task_id", payload.TaskId)
		return nil, err
	}
	n.logger.Debug("enqueue task into the queue system", "task_id", payload.TaskId)

	response.Status = avsproto.ExecutionStatus_EXECUTION_STATUS_PENDING
	return response, nil
}

// SimulateTask executes a complete task simulation by first running the trigger immediately,
// then executing the task nodes in sequence just like regular task execution.
// This is useful for testing tasks without waiting for actual trigger conditions.
// The task definition is provided in the request, so no storage persistence is required.
// SimulateTask runs a workflow without persisting it. The optional chainID
// argument (variadic for backward compatibility with existing callers) sets
// the workflow-level chain context; 0 or omitted means use the aggregator
// default. Per-node / per-trigger chain_id (set on the node configs) takes
// precedence over this value.
func (n *Engine) SimulateWorkflow(user *model.User, trigger *avsproto.TaskTrigger, nodes []*avsproto.TaskNode, edges []*avsproto.TaskEdge, inputVariables map[string]interface{}, chainIDs ...int64) (*avsproto.Execution, error) {
	return n.SimulateWorkflowWithContext(context.Background(), user, trigger, nodes, edges, inputVariables, chainIDs...)
}

// SimulateWorkflowWithContext is SimulateWorkflow with a caller-supplied
// context. The REST simulate handler passes the request context so a client
// disconnect / timeout cancels the worker-routed trigger reads (e.g. a
// BlockTrigger's GetBlockNumber / HeaderByNumber) reached via
// runTriggerImmediately.
func (n *Engine) SimulateWorkflowWithContext(ctx context.Context, user *model.User, trigger *avsproto.TaskTrigger, nodes []*avsproto.TaskNode, edges []*avsproto.TaskEdge, inputVariables map[string]interface{}, chainIDs ...int64) (*avsproto.Execution, error) {
	var chainID int64
	if len(chainIDs) > 0 {
		chainID = chainIDs[0]
	}
	// Validate inputVariables.settings (same requirements as CreateTask)
	if inputVariables == nil {
		return nil, status.Errorf(codes.InvalidArgument, "invalid task argument: inputVariables is required")
	}
	settingsIface, ok := inputVariables["settings"]
	if !ok {
		return nil, status.Errorf(codes.InvalidArgument, "invalid task argument: inputVariables.settings is required")
	}
	settings, ok := settingsIface.(map[string]interface{})
	if !ok {
		return nil, status.Errorf(codes.InvalidArgument, "invalid task argument: inputVariables.settings must be an object")
	}
	// settings.name is optional — simulations don't have a persistence
	// surface, and the REST mapper auto-mirrors CreateWorkflowRequest.name
	// into settings.name for the create path. An empty name here just
	// produces a transient simulation with no display label, which is
	// fine for what-if checks.
	taskName, _ := settings["name"].(string)
	smartWalletAddress, _ := settings["runner"].(string)
	if strings.TrimSpace(smartWalletAddress) == "" {
		return nil, status.Errorf(codes.InvalidArgument, "invalid task argument: inputVariables.settings.runner is required")
	}
	if !common.IsHexAddress(smartWalletAddress) {
		return nil, status.Errorf(codes.InvalidArgument, "invalid task argument: inputVariables.settings.runner must be a valid hex address")
	}

	// Note: SimulateTask intentionally skips ValidWalletOwner. Simulations are
	// allowed against any valid address so users can test workflows before
	// deploying smart wallets. Ownership is enforced only on CreateTask.

	// Create a temporary task structure for simulation (not saved to storage)
	simulationTaskID := model.GenerateID()

	task := &model.Workflow{
		Task: &avsproto.Task{
			Id:                 simulationTaskID,
			Name:               taskName,
			Owner:              user.Address.Hex(),
			SmartWalletAddress: smartWalletAddress,
			Trigger:            trigger,
			Nodes:              nodes,
			Edges:              edges,
			Status:             avsproto.TaskStatus_Enabled, // Set as enabled for simulation
		},
	}

	// Basic validation: Check if task structure is valid for execution
	if task.Trigger == nil {
		return nil, status.Errorf(codes.InvalidArgument, "task trigger is required for simulation")
	}
	if len(task.Nodes) == 0 {
		return nil, status.Errorf(codes.InvalidArgument, "task must have at least one node for simulation")
	}

	// Validate all node names for JavaScript compatibility
	if err := validateAllNodeNamesForJavaScript(task); err != nil {
		return nil, status.Errorf(codes.InvalidArgument, "node name validation failed: %v", err)
	}

	// Step 1: Simulate the trigger to get trigger output data
	// Extract trigger type and config from the TaskTrigger
	n.logger.Info("SimulateTask received trigger", "trigger_type_raw", trigger.GetType(), "trigger_type_int", int(trigger.GetType()), "trigger_id", trigger.GetId(), "trigger_name", trigger.GetName())

	// Debug: Check what oneof field is set
	n.logger.Info("SimulateTask trigger oneof debug",
		"fixed_time", trigger.GetFixedTime() != nil,
		"cron", trigger.GetCron() != nil,
		"block", trigger.GetBlock() != nil,
		"event", trigger.GetEvent() != nil,
		"manual", TaskTriggerToTriggerType(trigger) == avsproto.TriggerType_TRIGGER_TYPE_MANUAL,
		"oneof_type", fmt.Sprintf("%T", trigger.GetTriggerType()))

	// Use TaskTriggerToTriggerType to determine type from oneof field instead of just GetType()
	triggerType := TaskTriggerToTriggerType(trigger)
	n.logger.Info("SimulateTask trigger type conversion", "from_oneof", triggerType, "from_explicit", trigger.GetType())

	// Validate that the derived trigger type matches the expected type
	if triggerType != trigger.GetType() {
		n.logger.Error("Trigger type mismatch", "derived_type", triggerType, "expected_type", trigger.GetType(), "trigger_id", trigger.GetId(), "trigger_name", trigger.GetName())
		return nil, status.Errorf(codes.InvalidArgument, "trigger type mismatch: derived=%v, expected=%v", triggerType, trigger.GetType())
	}

	triggerTypeStr := TriggerTypeToString(triggerType)
	if triggerTypeStr == "" {
		return nil, status.Errorf(codes.InvalidArgument, "unsupported trigger type: %v (oneof type: %T)", trigger.GetType(), trigger.GetTriggerType())
	}

	// Extract trigger config using the shared utility function
	triggerConfig := TaskTriggerToConfig(trigger)

	// Validate manual trigger data if present (before execution)
	if triggerType == avsproto.TriggerType_TRIGGER_TYPE_MANUAL {
		if data, exists := triggerConfig["data"]; exists && data != nil {
			// Parse language from config (strict requirement - no default)
			lang, err := ParseLanguageFromConfig(triggerConfig)
			if err != nil {
				return nil, err
			}

			// Validate based on language using universal validator
			if err := ValidateInputByLanguage(data, lang); err != nil {
				return nil, err
			}
		}
	}

	// Step 1: Start timing BEFORE trigger execution (consistent with node timing)
	triggerStartTime := time.Now()

	triggerOutput, err := n.runTriggerImmediately(ctx, triggerTypeStr, triggerConfig, inputVariables)
	if err != nil {
		return nil, fmt.Errorf("failed to simulate trigger: %w", err)
	}

	// Step 2: Capture trigger end time AFTER trigger execution completes
	triggerEndTime := time.Now()

	// Step 3: Create QueueExecutionData similar to regular task execution
	simulationID := model.GenerateID()

	// Convert trigger output to proper protobuf structure using shared functions
	var triggerOutputProto interface{}
	switch triggerType {
	case avsproto.TriggerType_TRIGGER_TYPE_MANUAL:
		triggerOutputProto = buildManualTriggerOutput(triggerOutput)
	case avsproto.TriggerType_TRIGGER_TYPE_FIXED_TIME:
		triggerOutputProto = buildFixedTimeTriggerOutput(triggerOutput)
	case avsproto.TriggerType_TRIGGER_TYPE_CRON:
		triggerOutputProto = buildCronTriggerOutput(triggerOutput)
	case avsproto.TriggerType_TRIGGER_TYPE_BLOCK:
		triggerOutputProto = buildBlockTriggerOutput(triggerOutput)
	case avsproto.TriggerType_TRIGGER_TYPE_EVENT:
		triggerOutputProto = buildEventTriggerOutput(triggerOutput)
	default:
		return nil, fmt.Errorf("unsupported trigger type for simulation: %v", triggerType)
	}

	queueData := &QueueExecutionData{
		TriggerType:   trigger.Type,
		TriggerOutput: triggerOutputProto,
		ExecutionID:   simulationID,
	}

	// Step 4: Load secrets for the task
	secrets, err := LoadSecretForTask(n.db, task)
	if err != nil {
		n.logger.Warn("Failed to load secrets for workflow simulation", "error", err, "task_id", task.Id)
		// Don't fail the simulation, just use empty secrets
		secrets = make(map[string]string)
	}

	// Step 5: Create VM with simulated trigger data (similar to RunTask).
	// The simulation's default config chain comes from the optional chainID
	// arg (falling back to the aggregator default); per-node chains still win
	// via the chainConfigResolver wired below.
	simChainID := chainID
	if simChainID == 0 {
		simChainID = n.defaultChainID()
	}
	triggerReason := GetTriggerReasonOrDefault(queueData, task.Id, n.logger)
	vm, err := NewVMWithData(task, triggerReason, n.ResolveSmartWalletConfig(simChainID), secrets)
	if err != nil {
		return nil, fmt.Errorf("failed to create VM for simulation: %w", err)
	}
	vm.tenderlyClient = n.tenderlyClient

	vm.WithLogger(n.logger).WithDb(n.db).WithChainConfigResolver(n.ResolveSmartWalletConfig).SetSimulation(true)
	// Resolve AA sender for simulation ONLY if the workflow contains AA-relevant nodes
	// (contractWrite or ethTransfer, including loop nodes with these runners).
	// For non-AA workflows (e.g., CustomCode), skip this requirement.
	{
		requiresAA := taskNodesRequireAASender(task.Nodes)

		if requiresAA {
			owner := user.Address
			var chosenSender common.Address

			// Extract runner from settings (required for AA operations)
			var runnerStr string
			if settings, ok := inputVariables["settings"]; ok {
				if settingsMap, ok := settings.(map[string]interface{}); ok {
					if runnerIface, ok := settingsMap["runner"]; ok {
						if rs, ok := runnerIface.(string); ok && rs != "" {
							runnerStr = rs
						}
					}
				}
			}

			// Validate the runner against registered wallets
			if runnerStr != "" {
				resp, err := n.ListWallets(owner, &avsproto.ListWalletReq{})
				if err == nil {
					for _, w := range resp.GetItems() {
						if strings.EqualFold(w.GetAddress(), runnerStr) {
							chosenSender = common.HexToAddress(w.GetAddress())
							break
						}
					}
				}
			}

			// Fallbacks for simulation: if runner missing or not matched, use known smart wallet
			if (chosenSender == common.Address{}) {
				// Prefer the user's default smart account address when available
				if user.SmartAccountAddress != nil && (*user.SmartAccountAddress != common.Address{}) {
					chosenSender = *user.SmartAccountAddress
				} else {
					// As a last resort, pick the first wallet owned by the user (if any)
					if resp, err := n.ListWallets(owner, &avsproto.ListWalletReq{}); err == nil && len(resp.GetItems()) > 0 {
						chosenSender = common.HexToAddress(resp.GetItems()[0].GetAddress())
					}
				}
			}

			if (chosenSender == common.Address{}) {
				return nil, NewStructuredError(
					avsproto.ErrorCode_SMART_WALLET_NOT_FOUND,
					fmt.Sprintf("runner does not match any existing smart wallet for owner %s", owner.Hex()),
					map[string]interface{}{
						"owner": owner.Hex(),
					},
				)
			}

			// Set aa_sender for simulation writes to always use the runner (smart wallet)
			vm.AddVar("aa_sender", chosenSender.Hex())
			if n.logger != nil {
				n.logger.Info("SimulateTask: AA sender resolved", "sender", chosenSender.Hex())
			}
		} else if n.logger != nil {
			n.logger.Info("SimulateTask: Skipping AA sender resolution (no AA-relevant nodes in workflow)")
		}
	}

	// Add input variables to VM for template processing
	processedInputVariables := inputVariables
	for key, processedValue := range processedInputVariables {
		vm.AddVar(key, processedValue)
	}

	// Step 6: Add trigger data and input data together for JavaScript access
	// This ensures scripts can access both trigger.data and trigger.input
	// The buildTriggerDataMap function will handle EventTrigger data extraction internally
	triggerDataMap := buildTriggerDataMap(triggerReason.Type, triggerOutput)

	// Debug logging for EventTriggers to track data extraction
	if triggerReason.Type == avsproto.TriggerType_TRIGGER_TYPE_EVENT {
		n.logger.Debug("🔍 SimulateTask: EventTrigger data extraction", "inputKeys", GetMapKeys(triggerOutput), "outputKeys", GetMapKeys(triggerDataMap))
	}

	// Extract trigger config data if available
	triggerInputData := TaskTriggerToConfig(task.Trigger)

	// Build complete trigger variable data using shared function
	triggerVarData := buildTriggerVariableData(task.Trigger, triggerDataMap, triggerInputData)

	// Add the complete trigger variable with the actual trigger name for JavaScript access
	vm.AddVar(sanitizeTriggerNameForJS(trigger.GetName()), triggerVarData)

	// Step 7: Compile the workflow
	if err = vm.Compile(); err != nil {
		return nil, fmt.Errorf("failed to compile workflow for simulation: %w", err)
	}

	// Step 8: Create and add a trigger execution step with ACTUAL timing
	// Keys of the simulation input map (template variables), for execution step metadata.
	inputVariableKeys := make([]string, 0, len(inputVariables))
	for key := range inputVariables {
		inputVariableKeys = append(inputVariableKeys, key)
	}

	// Use the trigger config data for the execution step's Config field (includes data, headers, pathParams for ManualTrigger)
	// The Config field should show the configuration data used to execute the trigger
	var triggerConfigProto *structpb.Value
	triggerInputData = TaskTriggerToConfig(task.Trigger)
	if len(triggerInputData) > 0 {
		var err error
		triggerConfigProto, err = structpb.NewValue(triggerInputData)
		if err != nil {
			n.logger.Warn("Failed to convert trigger input to protobuf", "error", err)
			// Try a fallback approach: convert to JSON and back to ensure proper formatting
			jsonBytes, jsonErr := json.Marshal(triggerInputData)
			if jsonErr == nil {
				var cleanData interface{}
				if unmarshalErr := json.Unmarshal(jsonBytes, &cleanData); unmarshalErr == nil {
					if inputProto, err := structpb.NewValue(cleanData); err == nil {
						triggerConfigProto = inputProto
						n.logger.Info("✅ Successfully converted trigger input using JSON fallback")
					}
				}
			}
		}
	} else {
		n.logger.Info("🔍 SimulateTask: No trigger input found", "trigger_id", task.Trigger.Id, "trigger_type", task.Trigger.GetType())
	}

	// Use shared function to extract trigger status
	triggerSuccess, triggerError := extractTriggerStatus(triggerOutput)
	triggerLogMessage := fmt.Sprintf("Simulated trigger: %s executed successfully", task.Trigger.Name)
	if !triggerSuccess {
		triggerLogMessage = fmt.Sprintf("Simulated trigger: %s conditions not met", task.Trigger.Name)
	}

	triggerStep := &avsproto.Execution_Step{
		Id:      task.Trigger.Id, // Use new 'id' field
		Success: triggerSuccess,
		Error:   triggerError,
		StartAt: triggerStartTime.UnixMilli(), // Use actual trigger start time
		EndAt:   triggerEndTime.UnixMilli(),   // Use actual trigger end time
		Log:     triggerLogMessage,
		Inputs:  inputVariableKeys,              // Keys from simulation inputVariables map
		Type:    queueData.TriggerType.String(), // Use trigger type as string
		Name:    task.Trigger.Name,              // Use new 'name' field
		Config:  triggerConfigProto,             // Include trigger configuration data for debugging
	}

	// Attach execution_context on trigger step
	if vm != nil {
		provider := string(ProviderChainRPC)
		if vm.IsSimulation {
			provider = string(ProviderTenderly)
		}
		ctxMap := map[string]interface{}{
			"is_simulated": vm.IsSimulation,
			"provider":     provider,
		}
		if vm.smartWalletConfig != nil && vm.smartWalletConfig.ChainID != 0 {
			ctxMap["chain_id"] = vm.smartWalletConfig.ChainID
		}
		if ctxVal, err := structpb.NewValue(ctxMap); err == nil {
			triggerStep.ExecutionContext = ctxVal
		}
	}

	// Set trigger output data in the step using shared function
	triggerStep.OutputData = buildExecutionStepOutputData(queueData.TriggerType, triggerOutputProto)

	// Set metadata using shared function when conditions are not met
	if !triggerSuccess {
		if metadata := extractTriggerMetadata(triggerOutput); metadata != nil {
			if metadataProto, err := structpb.NewValue(metadata); err == nil {
				triggerStep.Metadata = metadataProto
			}
		}
	}

	// Add trigger step to execution logs
	vm.ExecutionLogs = append(vm.ExecutionLogs, triggerStep)

	// Step 9: Check trigger result to decide whether to continue workflow execution
	shouldContinue := n.shouldContinueWorkflowExecution(triggerOutput, triggerType)

	var runErr error
	var nodeEndTime time.Time

	if shouldContinue {
		// Inject simulated state overrides so downstream Tenderly simulations see
		// the balance changes implied by the trigger event.
		n.logger.Debug("SimulateTask: state override check",
			"triggerType", triggerType,
			"isEventTrigger", triggerType == avsproto.TriggerType_TRIGGER_TYPE_EVENT,
			"simulationStateNil", vm.simulationState == nil)
		if triggerType == avsproto.TriggerType_TRIGGER_TYPE_EVENT && vm.simulationState != nil {
			n.injectEventTriggerStateOverrides(triggerOutput, vm)
		}

		// Run the workflow nodes
		runErr = vm.Run()
		nodeEndTime = time.Now()
		n.logger.Info("✅ Workflow execution continued - trigger conditions met")
	} else {
		// Skip workflow execution - trigger conditions not met
		nodeEndTime = time.Now()
		n.logger.Info("🚫 Workflow execution skipped - trigger conditions not met")
	}

	// Step 10: Analyze execution results from all steps
	executionError, failedStepCount, resultStatus := vm.AnalyzeExecutionResult()

	// Step 11: Calculate total gas cost for the workflow

	// Create execution result with proper status/error analysis
	execution := &avsproto.Execution{
		Id:           simulationID,
		StartAt:      triggerStartTime.UnixMilli(),           // Start with trigger start time
		EndAt:        nodeEndTime.UnixMilli(),                // End with node completion time
		Status:       convertToExecutionStatus(resultStatus), // Use enum status instead of boolean
		Error:        executionError,                         // Comprehensive error message from failed steps
		Steps:        vm.ExecutionLogs,                       // Now contains both trigger and node steps (including failed ones)
		Index:        task.ExecutionCount,                    // Use current execution count for simulation (0-based)
		ExecutionFee: buildExecutionFee(n.config.FeeRates),
		Cogs:         buildCOGSFromSteps(vm.ExecutionLogs),
		ValueFee:     buildValueFee(task.Nodes, n.config.FeeRates),
	}

	// Log execution status based on result type
	switch resultStatus {
	case ExecutionSuccess:
		n.logger.Info("workflow simulation completed successfully", "task_id", task.Id, "simulation_id", simulationID, "steps", len(execution.Steps))
	case ExecutionFailed:
		// Clean up error message to avoid stack traces in logs
		cleanErrorMsg := executionError
		stackTraceRegex := regexp.MustCompile(`(?m)^\s*at .*$`)
		cleanErrorMsg = stackTraceRegex.ReplaceAllString(cleanErrorMsg, "")
		cleanErrorMsg = strings.TrimSpace(cleanErrorMsg)

		// User-workflow simulation failure: per-step errors are captured in the
		// persisted execution steps. Log summary at Warn so it stays out of Sentry
		// error alerts.
		n.logger.Warn("workflow simulation completed with failures",
			"error", cleanErrorMsg,
			"task_id", task.Id,
			"simulation_id", simulationID,
			"failed_steps", failedStepCount,
			"total_steps", len(vm.ExecutionLogs))
	}

	// Handle VM-level errors if they occurred
	if runErr != nil {
		n.logger.Error("workflow simulation had VM-level error", "vm_error", runErr, "task_id", task.Id, "simulation_id", simulationID)
		if execution.Error == "" {
			execution.Error = fmt.Sprintf("VM execution error: %s", runErr.Error())
		} else {
			execution.Error = fmt.Sprintf("VM execution error: %s (step analysis: %s)", runErr.Error(), execution.Error)
		}
		execution.Status = avsproto.ExecutionStatus_EXECUTION_STATUS_ERROR
	}

	return execution, nil
}

// injectEventTriggerStateOverrides examines the event trigger output and populates
// the VM's simulation state map with implied balance changes. For Transfer events this
// means injecting the token balance the smart wallet would have received; for ETH
// transfers it overrides the ETH balance.
func (n *Engine) injectEventTriggerStateOverrides(triggerOutput map[string]interface{}, vm *VM) {
	if vm.simulationState == nil {
		n.logger.Warn("injectEventTriggerStateOverrides: simulationState is nil, skipping")
		return
	}

	n.logger.Debug("injectEventTriggerStateOverrides: called",
		"triggerOutputKeys", GetMapKeys(triggerOutput))

	// Only process if there is parsed event data
	rawData := triggerOutput["data"]
	n.logger.Debug("injectEventTriggerStateOverrides: raw data field",
		"type", fmt.Sprintf("%T", rawData),
		"isNil", rawData == nil)

	data, ok := rawData.(map[string]interface{})
	if !ok || data == nil {
		// Flat structure — some paths put fields at the top level
		n.logger.Debug("injectEventTriggerStateOverrides: data field not a map, using triggerOutput directly")
		data = triggerOutput
	}

	eventName, _ := data["eventName"].(string)
	n.logger.Debug("injectEventTriggerStateOverrides: resolved eventName",
		"eventName", eventName,
		"dataKeys", GetMapKeys(data))

	switch eventName {
	case "Transfer":
		n.injectTransferEventState(data, vm)
	default:
		n.logger.Debug("No simulation state injection for event type",
			"eventName", eventName)
	}
}

// injectTransferEventState handles ERC20 Transfer events by computing the balance
// change implied by the simulated transfer and injecting it as a storage override.
func (n *Engine) injectTransferEventState(data map[string]interface{}, vm *VM) {
	n.logger.Info("injectTransferEventState: called",
		"dataKeys", GetMapKeys(data))

	tokenAddr, _ := data["contractAddress"].(string)
	walletAddr, _ := data["walletAddress"].(string)
	valueStr, _ := data["value"].(string)

	n.logger.Info("injectTransferEventState: extracted fields",
		"tokenAddr", tokenAddr,
		"walletAddr", walletAddr,
		"value", valueStr,
		"direction", data["direction"])

	if tokenAddr == "" || walletAddr == "" || valueStr == "" {
		n.logger.Warn("Transfer event missing fields for state injection",
			"tokenAddr", tokenAddr, "walletAddr", walletAddr, "value", valueStr)
		return
	}

	transferAmount, ok := new(big.Int).SetString(valueStr, 10)
	if !ok {
		n.logger.Warn("Failed to parse transfer amount for state injection", "value", valueStr)
		return
	}

	// Determine direction: if the wallet is the sender, balance decreases.
	// Event enrichment uses "sent"/"received"; older paths may use "outgoing"/"incoming".
	direction, _ := data["direction"].(string)
	switch strings.ToLower(direction) {
	case "outgoing", "sent":
		transferAmount = new(big.Int).Neg(transferAmount)
	}

	// Check if this is an ETH transfer (native currency, not ERC20)
	// ETH transfers have no token contract or a zero address
	tokenAddress := common.HexToAddress(tokenAddr)
	if tokenAddress == (common.Address{}) {
		// Native ETH transfer
		holderAddress := common.HexToAddress(walletAddr)
		if vm.smartWalletConfig != nil && vm.smartWalletConfig.EthRpcUrl != "" {
			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			if err := vm.simulationState.InjectETHBalanceChange(ctx, vm.smartWalletConfig, holderAddress, transferAmount); err != nil {
				n.logger.Warn("Failed to inject ETH balance override", "error", err)
			}
		}
		return
	}

	// ERC20 transfer
	if vm.smartWalletConfig == nil || vm.smartWalletConfig.EthRpcUrl == "" {
		n.logger.Warn("No smart wallet config for ERC20 balance injection, skipping")
		return
	}

	holderAddress := common.HexToAddress(walletAddr)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	if err := vm.simulationState.InjectERC20BalanceChange(ctx, vm.smartWalletConfig, tokenAddress, holderAddress, transferAmount); err != nil {
		n.logger.Warn("Failed to inject ERC20 balance override for simulation",
			"error", err,
			"token", tokenAddr,
			"holder", walletAddr,
			"amount", transferAmount.String())
	}
}

// shouldContinueWorkflowExecution checks trigger output to determine if workflow should continue
func (n *Engine) shouldContinueWorkflowExecution(triggerOutput map[string]interface{}, triggerType avsproto.TriggerType) bool {
	// For eventTrigger, check the success field
	if triggerType == avsproto.TriggerType_TRIGGER_TYPE_EVENT {
		if success, exists := triggerOutput["success"]; exists {
			if successBool, ok := success.(bool); ok {
				return successBool
			}
		}

		// Fallback: check legacy triggered field in metadata for backward compatibility
		if metadata, exists := triggerOutput["metadata"]; exists {
			if metadataMap, ok := metadata.(map[string]interface{}); ok {
				if triggered, exists := metadataMap["triggered"]; exists {
					if triggeredBool, ok := triggered.(bool); ok {
						return triggeredBool
					}
				}
			}
		}

		// Fallback: check legacy conditionsMet field for backward compatibility
		if conditionsMet, exists := triggerOutput["conditionsMet"]; exists {
			if conditionsMetBool, ok := conditionsMet.(bool); ok {
				return conditionsMetBool
			}
		}

		// If no success/triggered/conditionsMet field found, assume triggered (for historical events)
		return true
	}

	// For other trigger types (manual, cron, block, fixedTime), always continue
	return true
}

// List Execution for a given task id
func (n *Engine) ListExecutions(user *model.User, payload *avsproto.ListExecutionsReq) (*avsproto.ListExecutionsResp, error) {
	// Validate all tasks own by the caller, if there are any tasks won't be owned by caller, we return permission error
	// Admin users (authenticated with API key) use zero address and can access any task
	isAdminUser := user.Address == (common.Address{})
	tasks := make(map[string]*model.Workflow)

	for _, id := range payload.TaskIds {
		task, err := n.GetWorkflowByID(id)
		if err != nil {
			return nil, status.Errorf(codes.NotFound, TaskNotFoundError)
		}

		if !isAdminUser && !task.OwnedBy(user.Address) {
			return nil, status.Errorf(codes.NotFound, TaskNotFoundError)
		}
		tasks[id] = task
	}

	// Build per-task, per-chain prefixes — chain-scoped history keys share
	// no common prefix below "history:" so we must enumerate every chain
	// for each task.
	prefixes := make([]string, 0, len(payload.TaskIds)*len(n.knownChainIDs()))
	for _, id := range payload.TaskIds {
		prefixes = append(prefixes, n.taskExecutionPrefixes(id)...)
	}

	if n.logger != nil {
		n.logger.Info("ListExecutions: scanning prefixes", "count", len(prefixes), "prefixes", prefixes)
	}
	executionKeys, err := n.db.ListKeysMulti(prefixes)
	if n.logger != nil {
		n.logger.Info("ListExecutions: found execution keys", "count", len(executionKeys))
	}

	// Fallback: if no keys found using ListKeysMulti, try value scan per prefix
	if len(executionKeys) == 0 {
		for _, id := range payload.TaskIds {
			for _, prefix := range n.taskExecutionPrefixesBytes(id) {
				items, getErr := n.db.GetByPrefix(prefix)
				if getErr != nil {
					continue
				}
				for _, kv := range items {
					executionKeys = append(executionKeys, string(kv.Key))
				}
			}
		}
		if n.logger != nil {
			n.logger.Info("ListExecutions: fallback GetByPrefix yielded keys", "count", len(executionKeys))
		}
	}

	// second, do the sort, this is key sorted by ordering of their insertion
	sort.Slice(executionKeys, func(i, j int) bool {
		id1 := ulid.MustParse(string(ExecutionIdFromStorageKey([]byte(executionKeys[i]))))
		id2 := ulid.MustParse(string(ExecutionIdFromStorageKey([]byte(executionKeys[j]))))
		return id1.Compare(id2) < 0
	})

	if err != nil {
		return nil, status.Errorf(codes.Code(avsproto.ErrorCode_STORAGE_UNAVAILABLE), StorageUnavailableError)
	}

	executioResp := &avsproto.ListExecutionsResp{
		Items: []*avsproto.Execution{},
		PageInfo: &avsproto.PageInfo{
			StartCursor:     "",
			EndCursor:       "",
			HasPreviousPage: false,
			HasNextPage:     false,
		},
	}

	var before, after string
	var limitVal int64

	if payload != nil {
		before = payload.Before
		after = payload.After
		limitVal = payload.Limit
	}

	cursor, limit, err := SetupPagination(before, after, limitVal)
	if err != nil {
		return nil, err
	}

	total := 0
	var firstExecutionId, lastExecutionId string
	var hasMoreItems bool

	if !cursor.IsZero() && cursor.Direction == CursorDirectionPrevious {
		// Backward pagination: iterate oldest-to-newest to find items
		// immediately newer than the cursor, then reverse for display order.
		cursorUlid, err := ulid.Parse(cursor.Position)
		if err != nil {
			return nil, status.Errorf(codes.InvalidArgument, InvalidCursor)
		}
		for i := 0; i < len(executionKeys); i++ {
			key := executionKeys[i]
			executionUlid := ulid.MustParse(ExecutionIdFromStorageKey([]byte(key)))

			// Skip items at or before the cursor position
			if executionUlid.Compare(cursorUlid) <= 0 {
				continue
			}

			executionValue, err := n.db.GetKey([]byte(key))
			if err != nil {
				continue
			}

			exec := &avsproto.Execution{}
			// DiscardUnknown: tolerate proto fields renamed/removed
			// since this execution was written. The merge tool re-
			// serializes bodies on import so this matters mainly for
			// data that was never touched by a merge run.
			if err := (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(executionValue, exec); err == nil {
				executioResp.Items = append(executioResp.Items, exec)
				total += 1
			}
			// Collect one extra to detect if more items exist
			if total > limit {
				break
			}
		}
		// If we collected more than limit, trim the extra and mark hasMoreItems
		if len(executioResp.Items) > limit {
			executioResp.Items = executioResp.Items[:limit]
			hasMoreItems = true
		}
		// Reverse to newest-first display order
		for i, j := 0, len(executioResp.Items)-1; i < j; i, j = i+1, j-1 {
			executioResp.Items[i], executioResp.Items[j] = executioResp.Items[j], executioResp.Items[i]
		}
	} else {
		// Forward pagination (or no cursor): iterate newest-to-oldest
		for i := len(executionKeys) - 1; i >= 0; i-- {
			key := executionKeys[i]

			executionUlid := ulid.MustParse(ExecutionIdFromStorageKey([]byte(key)))
			if !cursor.IsZero() {
				if cursor.LessThanOrEqualUlid(executionUlid) {
					continue
				}
			}

			executionValue, err := n.db.GetKey([]byte(key))
			if err != nil {
				continue
			}

			exec := &avsproto.Execution{}
			// DiscardUnknown: tolerate proto fields renamed/removed
			// since this execution was written. The merge tool re-
			// serializes bodies on import so this matters mainly for
			// data that was never touched by a merge run.
			if err := (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(executionValue, exec); err == nil {
				executioResp.Items = append(executioResp.Items, exec)
				total += 1
			}
			// Collect one extra to detect if more items exist
			if total > limit {
				break
			}
		}
		// If we collected more than limit, trim the extra and mark hasMoreItems
		if len(executioResp.Items) > limit {
			executioResp.Items = executioResp.Items[:limit]
			hasMoreItems = true
		}
	}

	// Set pagination info
	if len(executioResp.Items) > 0 {
		firstExecutionId = executioResp.Items[0].Id
		lastExecutionId = executioResp.Items[len(executioResp.Items)-1].Id

		executioResp.PageInfo.StartCursor = CreateNextCursor(firstExecutionId)
		executioResp.PageInfo.EndCursor = CreateNextCursor(lastExecutionId)

		if !cursor.IsZero() && cursor.Direction == CursorDirectionPrevious {
			// We reached this page via backward navigation, so the cursor position
			// and everything older always forms at least one next page.
			executioResp.PageInfo.HasNextPage = true
			executioResp.PageInfo.HasPreviousPage = hasMoreItems
		} else {
			executioResp.PageInfo.HasNextPage = hasMoreItems
			executioResp.PageInfo.HasPreviousPage = !cursor.IsZero()
		}
	}
	return executioResp, nil
}

func (n *Engine) setExecutionStatusQueue(task *model.Workflow, executionID string) error {
	status := strconv.Itoa(int(avsproto.ExecutionStatus_EXECUTION_STATUS_PENDING))
	return n.db.Set(TaskTriggerKey(task, executionID), []byte(status))
}

func (n *Engine) getExecutionStatusFromQueue(task *model.Workflow, executionID string) (*avsproto.ExecutionStatus, error) {
	status, err := n.db.GetKey(TaskTriggerKey(task, executionID))
	if err != nil {
		return nil, err
	}

	value, err := strconv.Atoi(string(status))
	if err != nil {
		return nil, err
	}
	statusValue := avsproto.ExecutionStatus(value)
	return &statusValue, nil
}

// SignalExecution delivers an external/approval signal to a WAITING execution
// (durable execution — the human-approval transport). The authenticated user must
// own the workflow (GetWorkflow → NotFound otherwise); the decision is validated
// (InvalidArgument); and a DeliverSignal gate failure (no pending wait / mismatched
// kind / timed out) is mapped to FailedPrecondition, so the REST layer returns a
// 4xx rather than a 500. Decision is "approve" | "reject".
func (n *Engine) SignalExecution(user *model.User, workflowID, executionID, decision string, payload *structpb.Value) (*avsproto.Execution, error) {
	if decision != "approve" && decision != "reject" {
		return nil, status.Errorf(codes.InvalidArgument, "decision must be 'approve' or 'reject', got %q", decision)
	}
	task, err := n.GetWorkflow(user, workflowID)
	if err != nil {
		return nil, err // GetWorkflow returns codes.NotFound for a missing/non-owned workflow
	}
	// Serialize concurrent resumes of this execution (exactly-once — see executionMutex).
	mu := n.executionMutex(executionID)
	mu.Lock()
	defer mu.Unlock()
	executor := NewExecutor(n.ResolveSmartWalletConfig(n.defaultChainID()), n.db, n.logger, n, n.priceService)
	exec, err := executor.DeliverSignal(task, &Signal{
		ExecutionID: executionID,
		Kind:        WakeExternalSignal,
		Decision:    decision,
		Approver:    strings.ToLower(user.Address.Hex()),
		Payload:     payload,
	})
	if err != nil {
		return nil, status.Errorf(codes.FailedPrecondition, "deliver signal: %v", err)
	}
	return exec, nil
}

// GetExecution for a given task id and execution id
func (n *Engine) GetExecution(user *model.User, payload *avsproto.ExecutionReq) (*avsproto.Execution, error) {
	task, err := n.GetWorkflow(user, payload.TaskId)
	if err != nil {
		return nil, err
	}

	// First try to get completed execution from storage
	rawExecution, err := n.db.GetKey(TaskExecutionKey(task, payload.ExecutionId))
	if err == nil {
		exec := &avsproto.Execution{}
		// DiscardUnknown: tolerate proto fields renamed/removed since
		// the execution was written; otherwise GetExecution/
		// GetExecutionStatus would error 500 on any older row.
		err = (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(rawExecution, exec)
		if err != nil {
			return nil, status.Errorf(codes.Code(avsproto.ErrorCode_TASK_DATA_CORRUPTED), TaskStorageCorruptedError)
		}
		return exec, nil
	}

	// If not found in completed executions, check if it's pending in queue
	execStatus, err := n.getExecutionStatusFromQueue(task, payload.ExecutionId)
	if err != nil {
		return nil, status.Errorf(codes.NotFound, ExecutionNotFoundError)
	}

	// Create a minimal execution object for pending status
	// We don't have full execution details yet, but we can return basic info

	// For pending executions created by non-blocking triggers, try to get the pre-assigned index
	// from the pending execution storage
	pendingIndex := int64(0) // Default to 0 if not found
	pendingKey := PendingExecutionKey(task, payload.ExecutionId)
	if pendingData, err := n.db.GetKey(pendingKey); err == nil {
		if storedIndex, parseErr := strconv.ParseInt(string(pendingData), 10, 64); parseErr == nil {
			pendingIndex = storedIndex
			n.logger.Debug("🔍 Retrieved pre-assigned index from pending storage", "task_id", payload.TaskId, "execution_id", payload.ExecutionId, "index", pendingIndex)
		} else {
			n.logger.Debug("Pending data not an index, using fallback", "task_id", payload.TaskId, "execution_id", payload.ExecutionId, "pending_data", string(pendingData))
			// Fallback: assign new atomic index for pending executions without stored index
			if atomicIndex, indexErr := n.AssignNextExecutionIndex(task); indexErr == nil {
				pendingIndex = atomicIndex
				n.logger.Debug("Assigned new atomic index for pending execution", "task_id", payload.TaskId, "execution_id", payload.ExecutionId, "index", pendingIndex)
			}
		}
	} else {
		// No pending data found, assign new atomic index
		if atomicIndex, indexErr := n.AssignNextExecutionIndex(task); indexErr == nil {
			pendingIndex = atomicIndex
			n.logger.Debug("Assigned new atomic index for pending execution (no stored index)", "task_id", payload.TaskId, "execution_id", payload.ExecutionId, "index", pendingIndex)
		}
	}

	n.logger.Debug("🗂️ Returning pending execution", "task_id", payload.TaskId, "execution_id", payload.ExecutionId, "status", *execStatus, "index", pendingIndex)
	return &avsproto.Execution{
		Id:           payload.ExecutionId,
		Status:       *execStatus,
		StartAt:      time.Now().UnixMilli(),               // Approximate start time
		Steps:        []*avsproto.Execution_Step{},         // Empty steps for pending
		Index:        pendingIndex,                         // Use pre-assigned or newly assigned index
		ExecutionFee: buildExecutionFee(n.config.FeeRates), // Known upfront
		Cogs:         []*avsproto.NodeCOGS{},               // Unknown until execution completes
		ValueFee:     nil,                                  // Unknown until execution completes
	}, nil
}

func (n *Engine) GetExecutionStatus(user *model.User, payload *avsproto.ExecutionReq) (*avsproto.ExecutionStatusResp, error) {
	task, err := n.GetWorkflow(user, payload.TaskId)
	if err != nil {
		return nil, err
	}

	// First check if execution is completed and stored
	rawExecution, err := n.db.GetKey(TaskExecutionKey(task, payload.ExecutionId))
	if err == nil {
		exec := &avsproto.Execution{}
		// DiscardUnknown: tolerate proto fields renamed/removed since
		// the execution was written; otherwise GetExecution/
		// GetExecutionStatus would error 500 on any older row.
		err = (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(rawExecution, exec)
		if err != nil {
			return nil, status.Errorf(codes.Code(avsproto.ErrorCode_TASK_DATA_CORRUPTED), TaskStorageCorruptedError)
		}

		// Return the execution status directly from the stored execution
		// The status field now contains the proper enum value from AnalyzeExecutionResult
		return &avsproto.ExecutionStatusResp{Status: exec.Status}, nil
	}

	// Check if it's pending in queue
	execStatus, err := n.getExecutionStatusFromQueue(task, payload.ExecutionId)
	if err != nil {
		return nil, status.Errorf(codes.NotFound, ExecutionNotFoundError)
	}

	return &avsproto.ExecutionStatusResp{Status: *execStatus}, nil
}

func (n *Engine) GetExecutionCount(user *model.User, payload *avsproto.GetExecutionCountReq) (*avsproto.GetExecutionCountResp, error) {
	workflowIds := payload.WorkflowIds

	total := int64(0)
	var err error

	if len(workflowIds) == 0 {
		workflowIds = []string{}
		// count all executions of the owner by finding all their task idds
		var taskIds [][]byte
		var lastErr error
		for _, prefix := range n.chainUserPrefixesBytes(user.Address) {
			chunk, e := n.db.GetKeyHasPrefix(prefix)
			if e != nil {
				lastErr = e
				continue
			}
			taskIds = append(taskIds, chunk...)
		}
		err := lastErr
		_ = err // legacy variable kept; check below operates on err

		if err != nil {
			return nil, status.Errorf(codes.Internal, "Internal error counting execution")
		}
		for _, id := range taskIds {
			taskId := TaskIdFromTaskStatusStorageKey(id)
			workflowIds = append(workflowIds, string(taskId))
		}
	}

	prefixes := [][]byte{}
	for _, id := range workflowIds {
		if len(id) != TaskIDLength {
			continue
		}
		prefixes = append(prefixes, n.taskExecutionPrefixesBytes(id)...)
	}
	total, err = n.db.CountKeysByPrefixes(prefixes)

	if err != nil {
		n.logger.Error("error counting execution for", "user", user.Address, "error", err)
		return nil, status.Errorf(codes.Internal, "Internal error counting execution")
	}

	return &avsproto.GetExecutionCountResp{
		Total: total,
	}, nil
}

func (n *Engine) DeleteWorkflowByUser(user *model.User, taskID string) (*avsproto.DeleteTaskResp, error) {
	n.logger.Info("🔄 Starting delete task operation", "task_id", taskID, "user", user.Address.String())

	task, err := n.GetWorkflow(user, taskID)
	if err != nil {
		n.logger.Warn("❌ Task not found for deletion", "task_id", taskID, "error", err)
		return &avsproto.DeleteTaskResp{
			Success: false,
			Status:  "not_found",
			Message: "Task not found",
			Id:      taskID,
		}, nil
	}

	n.logger.Info("✅ Retrieved task for deletion", "task_id", taskID, "status", task.Status)

	if task.Status == avsproto.TaskStatus_Running {
		n.logger.Warn("❌ Cannot delete executing task", "task_id", taskID, "status", task.Status)
		return &avsproto.DeleteTaskResp{
			Success:        false,
			Status:         "cannot_delete",
			Message:        "Only non executing task can be deleted",
			Id:             taskID,
			PreviousStatus: getTaskStatusString(task.Status),
		}, nil
	}

	previousStatus := getTaskStatusString(task.Status)
	deletedAt := time.Now().UnixMilli()

	n.logger.Info("🗑️ Deleting task storage", "task_id", taskID)
	if err := n.db.Delete(WorkflowStorageKey(task.Id, task.Status)); err != nil {
		n.logger.Error("failed to delete task storage", "error", err, "task_id", task.Id)
		return &avsproto.DeleteTaskResp{
			Success: false,
			Status:  "error",
			Message: fmt.Sprintf("Failed to delete task: %v", err),
			Id:      taskID,
		}, nil
	}

	n.logger.Info("🗑️ Deleting task user key", "task_id", taskID)
	if err := n.db.Delete(TaskUserKey(task)); err != nil {
		n.logger.Error("failed to delete task user key", "error", err, "task_id", task.Id)
		return &avsproto.DeleteTaskResp{
			Success: false,
			Status:  "error",
			Message: fmt.Sprintf("Failed to delete task user key: %v", err),
			Id:      taskID,
		}, nil
	}

	// GC durable state of any suspended (WAITING) execution this task owned, so its
	// checkpoint/wake — and any operator internal trigger — don't outlive the workflow.
	n.gcDurableStateForTask(taskID)

	// Cascade-delete this workflow's durable per-run state ({{state.*}} /
	// wfstate:<taskId>) so it doesn't outlive the task (scoped by taskID, so this
	// touches only this workflow's namespace).
	if stateKeys, listErr := n.db.ListKeys(string(WorkflowStatePrefix(taskID))); listErr == nil {
		for _, k := range stateKeys {
			if delErr := n.db.Delete([]byte(k)); delErr != nil {
				n.logger.Warn("failed to delete workflow state key", "key", k, "error", delErr)
			}
		}
	} else {
		n.logger.Warn("failed to list workflow state keys for cascade delete; state may be orphaned", "task_id", taskID, "error", listErr)
	}

	n.logger.Info("📢 Starting operator notifications", "task_id", taskID)
	n.notifyOperatorsTaskOperation(taskID, avsproto.MessageOp_DeleteTask)
	n.logger.Info("✅ Delete task operation completed", "task_id", taskID)

	return &avsproto.DeleteTaskResp{
		Success:        true,
		Status:         "deleted",
		Message:        "Task deleted successfully",
		DeletedAt:      deletedAt,
		Id:             taskID,
		PreviousStatus: previousStatus,
	}, nil
}

func (n *Engine) SetWorkflowEnabledByUser(user *model.User, taskID string, enabled bool) (*avsproto.SetTaskEnabledResp, error) {
	task, err := n.GetWorkflow(user, taskID)
	if err != nil {
		return &avsproto.SetTaskEnabledResp{
			Success: false,
			Status:  "not_found",
			Message: "Task not found",
			Id:      taskID,
		}, nil
	}

	oldStatus := task.Status

	// Terminal states cannot be toggled
	if oldStatus == avsproto.TaskStatus_Completed || oldStatus == avsproto.TaskStatus_Failed {
		return &avsproto.SetTaskEnabledResp{
			Success:        false,
			Status:         "error",
			Message:        fmt.Sprintf("Cannot toggle task from terminal status: %s", getTaskStatusString(oldStatus)),
			Id:             taskID,
			PreviousStatus: getTaskStatusString(oldStatus),
		}, nil
	}

	// Running guard: do not alter status; treat active=true as idempotent and active=false as disallowed
	if oldStatus == avsproto.TaskStatus_Running {
		if enabled {
			return &avsproto.SetTaskEnabledResp{
				Success:        true,
				Status:         "enabled",
				Message:        "Task is already running",
				Id:             taskID,
				PreviousStatus: getTaskStatusString(oldStatus),
				UpdatedAt:      time.Now().UnixMilli(),
			}, nil
		}
		return &avsproto.SetTaskEnabledResp{
			Success:        false,
			Status:         "error",
			Message:        "Cannot disable a running task",
			Id:             taskID,
			PreviousStatus: getTaskStatusString(oldStatus),
		}, nil
	}

	// Idempotent responses
	if enabled && oldStatus == avsproto.TaskStatus_Enabled {
		return &avsproto.SetTaskEnabledResp{
			Success:        true,
			Status:         "enabled",
			Message:        "Task is already enabled",
			Id:             taskID,
			PreviousStatus: getTaskStatusString(oldStatus),
			UpdatedAt:      time.Now().UnixMilli(),
		}, nil
	}
	if !enabled && oldStatus == avsproto.TaskStatus_Disabled {
		return &avsproto.SetTaskEnabledResp{
			Success:        true,
			Status:         "disabled",
			Message:        "Task is already disabled",
			Id:             taskID,
			PreviousStatus: getTaskStatusString(oldStatus),
			UpdatedAt:      time.Now().UnixMilli(),
		}, nil
	}

	updates := map[string][]byte{}

	if enabled {
		task.SetEnabled()
	} else {
		task.SetDisabled()
	}

	// Serialize and persist
	taskJSON, err := task.ToJSON()
	if err != nil {
		return &avsproto.SetTaskEnabledResp{
			Success:        false,
			Status:         "error",
			Message:        fmt.Sprintf("Failed to serialize task: %v", err),
			Id:             taskID,
			PreviousStatus: getTaskStatusString(oldStatus),
		}, nil
	}
	updates[string(WorkflowStorageKey(task.Id, task.Status))] = taskJSON
	updates[string(TaskUserKey(task))] = []byte(fmt.Sprintf("%d", task.Status))

	if err = n.db.BatchWrite(updates); err != nil {
		return &avsproto.SetTaskEnabledResp{
			Success:        false,
			Status:         "error",
			Message:        fmt.Sprintf("Failed to update task status: %v", err),
			Id:             taskID,
			PreviousStatus: getTaskStatusString(oldStatus),
		}, nil
	}

	// Delete old record if different status
	if oldStatus != task.Status {
		if delErr := n.db.Delete(WorkflowStorageKey(task.Id, oldStatus)); delErr != nil {
			n.logger.Error("failed to delete old task status entry", "error", delErr, "task_id", task.Id, "old_status", oldStatus)
		}
	}

	// Update in-memory tracking and notify operators
	if enabled {
		n.lock.Lock()
		n.tasks[task.Id] = task
		n.lock.Unlock()
		// Notify operators to start monitoring the task immediately
		// instead of waiting for the next StreamCheckToOperator ticker cycle
		n.sendMonitorTaskTriggerToOperators(task)
	} else {
		n.lock.Lock()
		delete(n.tasks, task.Id)
		n.lock.Unlock()
		// Notify operators to remove checks immediately
		n.notifyOperatorsTaskOperation(taskID, avsproto.MessageOp_DisableTask)
	}

	statusStr := "disabled"
	messageStr := "Task disabled successfully"
	if enabled {
		statusStr = "enabled"
		messageStr = "Task enabled successfully"
	}

	return &avsproto.SetTaskEnabledResp{
		Success:        true,
		Status:         statusStr,
		Message:        messageStr,
		Id:             taskID,
		PreviousStatus: getTaskStatusString(oldStatus),
		UpdatedAt:      time.Now().UnixMilli(),
	}, nil
}

// DisableTask turns off a task without user authentication (for internal use like overload alerts)
func (n *Engine) DisableWorkflow(taskID string) (bool, error) {
	n.lock.Lock()
	task, exists := n.tasks[taskID]
	n.lock.Unlock()

	if !exists {
		n.logger.Warn("Task not found for disabling", "task_id", taskID)
		return false, nil
	}

	if task.Status != avsproto.TaskStatus_Enabled {
		n.logger.Info("Task is not enabled, cannot disable", "task_id", taskID, "status", task.Status)
		return false, nil
	}

	updates := map[string][]byte{}
	oldStatus := task.Status
	task.SetDisabled()

	taskJSON, err := task.ToJSON()
	if err != nil {
		return false, fmt.Errorf("failed to serialize task during disabling: %w", err)
	}

	updates[string(WorkflowStorageKey(task.Id, task.Status))] = taskJSON
	updates[string(TaskUserKey(task))] = []byte(fmt.Sprintf("%d", task.Status))

	if err = n.db.BatchWrite(updates); err == nil {
		// Delete the old record
		if oldStatus != task.Status {
			if delErr := n.db.Delete(WorkflowStorageKey(task.Id, oldStatus)); delErr != nil {
				n.logger.Error("failed to delete old task status entry", "error", delErr, "task_id", task.Id, "old_status", oldStatus)
			}
		}

		n.lock.Lock()
		delete(n.tasks, task.Id) // Remove from active tasks map
		n.lock.Unlock()

	} else {
		return false, err
	}

	n.notifyOperatorsTaskOperation(taskID, avsproto.MessageOp_DisableTask)
	n.logger.Info("Task disabled due to system alert", "task_id", taskID)

	return true, nil
}

func (n *Engine) CreateSecret(user *model.User, payload *avsproto.CreateOrUpdateSecretReq) (bool, error) {
	secret := &model.Secret{
		User:       user,
		Name:       payload.Name,
		Value:      payload.Secret,
		OrgID:      payload.OrgId,
		WorkflowID: payload.WorkflowId,
	}

	updates := map[string][]byte{}
	if strings.HasPrefix(strings.ToLower(payload.Name), "ap_") {
		return false, status.Errorf(codes.InvalidArgument, "secret name cannot start with ap_")
	}

	if len(payload.Name) == 0 || len(payload.Name) > MaxSecretNameLength {
		return false, status.Errorf(codes.InvalidArgument, "secret name length is invalid: should be 1-255 character")
	}

	key, _ := SecretStorageKey(secret)
	updates[key] = []byte(payload.Secret)
	err := n.db.BatchWrite(updates)
	if err == nil {
		return true, nil
	}

	return false, status.Errorf(codes.Internal, "Cannot save data")
}

func (n *Engine) UpdateSecret(user *model.User, payload *avsproto.CreateOrUpdateSecretReq) (bool, error) {
	updates := map[string][]byte{}
	secret := &model.Secret{
		User:       user,
		Name:       payload.Name,
		Value:      payload.Secret,
		OrgID:      payload.OrgId,
		WorkflowID: payload.WorkflowId,
	}
	key, _ := SecretStorageKey(secret)
	if ok, err := n.db.Exist([]byte(key)); !ok || err != nil {
		return false, status.Errorf(codes.NotFound, "Secret not found")
	}

	updates[key] = []byte(payload.Secret)

	err := n.db.BatchWrite(updates)
	if err == nil {
		return true, nil
	}

	// In original code, it returned true, nil even on error. Preserving that.
	return true, nil
}

// ListSecrets
func (n *Engine) ListSecrets(user *model.User, payload *avsproto.ListSecretsReq) (*avsproto.ListSecretsResp, error) {
	prefixes := []string{
		SecretStoragePrefix(user),
	}

	result := &avsproto.ListSecretsResp{
		Items: []*avsproto.Secret{},
		PageInfo: &avsproto.PageInfo{
			StartCursor:     "",
			EndCursor:       "",
			HasPreviousPage: false,
			HasNextPage:     false,
		},
	}

	secretKeys, err := n.db.ListKeysMulti(prefixes)
	if err != nil {
		return nil, err
	}

	sort.Strings(secretKeys)

	var before, after string
	var limitVal int64

	if payload != nil {
		before = payload.Before
		after = payload.After
		limitVal = payload.Limit
	}

	cursor, limit, err := SetupPagination(before, after, limitVal)
	if err != nil {
		return nil, err
	}

	total := 0
	var firstKey, lastKey string
	var hasMoreItems bool
	var processedCount int

	// Process keys that match the cursor criteria and stop when limit+1 is reached
	// We fetch limit+1 to determine if there are more pages without loading everything
	for _, k := range secretKeys {
		if !cursor.IsZero() {
			if (cursor.Direction == CursorDirectionNext && k <= cursor.Position) ||
				(cursor.Direction == CursorDirectionPrevious && k >= cursor.Position) {
				continue
			}
		}

		processedCount++

		// If we've processed more than the limit, we know there are more items
		if processedCount > limit {
			hasMoreItems = true
			break
		}

		secretWithNameOnly := SecretNameFromKey(k)
		item := &avsproto.Secret{
			Name:       secretWithNameOnly.Name,
			OrgId:      secretWithNameOnly.OrgID,
			WorkflowId: secretWithNameOnly.WorkflowID,
			// Always include scope for basic functionality
			Scope: "user", // Default scope, could be enhanced to read from storage
		}

		// Conditionally populate additional fields based on request parameters
		if payload != nil {
			if payload.IncludeTimestamps {
				// In a real implementation, these would be fetched from storage
				// For now, we'll skip timestamps since we don't have real data
				// item.CreatedAt = time.Now().Unix() // Would fetch from storage
				// item.UpdatedAt = time.Now().Unix() // Would fetch from storage
			}
			if payload.IncludeCreatedBy {
				item.CreatedBy = user.Address.Hex() // Would fetch from storage
			}
			if payload.IncludeDescription {
				item.Description = "" // Would fetch from storage
			}
		}

		result.Items = append(result.Items, item)

		if total == 0 {
			firstKey = k
		}
		lastKey = k
		total++
	}

	// Set pagination info
	if len(result.Items) > 0 {
		// Always set cursors for the current page (GraphQL PageInfo convention)
		result.PageInfo.StartCursor = CreateNextCursor(firstKey)
		result.PageInfo.EndCursor = CreateNextCursor(lastKey)

		// Check if there are more items after the current page
		result.PageInfo.HasNextPage = hasMoreItems

		// Check if there are items before the current page
		// This is true if we have a cursor and we're not at the beginning
		result.PageInfo.HasPreviousPage = !cursor.IsZero() && cursor.Direction == CursorDirectionNext

		// For backward pagination, we need to check if there are items after
		if cursor.Direction == CursorDirectionPrevious {
			result.PageInfo.HasNextPage = true // There are items after since we're going backwards
			// Check if there are more items before
			result.PageInfo.HasPreviousPage = hasMoreItems
		}
	}

	return result, nil
}

func (n *Engine) DeleteSecret(user *model.User, payload *avsproto.DeleteSecretReq) (*avsproto.DeleteSecretResp, error) {
	// No need to check permission, the key is prefixed by user eoa already
	secret := &model.Secret{
		Name:       payload.Name,
		User:       user,
		OrgID:      payload.OrgId,
		WorkflowID: payload.WorkflowId,
	}
	key, _ := SecretStorageKey(secret)

	// Check if secret exists before attempting to delete
	exists, err := n.db.Exist([]byte(key))
	if err != nil {
		return &avsproto.DeleteSecretResp{
			Success:    false,
			Status:     "error",
			Message:    fmt.Sprintf("Error checking secret existence: %v", err),
			SecretName: payload.Name,
		}, err
	}

	// If the exact key wasn't found and no specific scope was requested,
	// scan the user's secrets to find a matching key by name.
	// This handles the case where a workflow-scoped secret is deleted
	// without passing the workflowId (e.g. from the Secret Settings page).
	if !exists && payload.WorkflowId == "" && payload.OrgId == "" {
		foundKey, foundScope := n.findSecretKeyByName(user, payload.Name)
		if foundKey != "" {
			key = foundKey
			exists = true
			// Update scope info from the found key
			if foundScope != nil {
				payload.WorkflowId = foundScope.WorkflowID
				payload.OrgId = foundScope.OrgID
			}
		}
	}

	if !exists {
		return &avsproto.DeleteSecretResp{
			Success:    true,
			Status:     "not_found",
			Message:    "Secret not found",
			SecretName: payload.Name,
		}, nil
	}

	// Attempt to delete the secret
	err = n.db.Delete([]byte(key))
	if err != nil {
		return &avsproto.DeleteSecretResp{
			Success:    false,
			Status:     "error",
			Message:    fmt.Sprintf("Error deleting secret: %v", err),
			SecretName: payload.Name,
		}, err
	}

	// Determine scope for response
	scope := "user"
	if payload.OrgId != "" {
		scope = "org"
	} else if payload.WorkflowId != "" {
		scope = "workflow"
	}

	return &avsproto.DeleteSecretResp{
		Success:    true,
		Status:     "deleted",
		Message:    "Secret successfully deleted",
		DeletedAt:  time.Now().UnixMilli(),
		SecretName: payload.Name,
		Scope:      scope,
	}, nil
}

// findSecretKeyByName scans the user's secret keys to find a secret by name,
// regardless of scope. Returns the full storage key and the parsed secret metadata.
func (n *Engine) findSecretKeyByName(user *model.User, name string) (string, *model.Secret) {
	prefixes := []string{
		SecretStoragePrefix(user),
	}

	secretKeys, err := n.db.ListKeysMulti(prefixes)
	if err != nil {
		return "", nil
	}

	var matches []string
	var matchMeta *model.Secret
	for _, k := range secretKeys {
		secretMeta := SecretNameFromKey(k)
		if secretMeta.Name == name {
			matches = append(matches, k)
			matchMeta = secretMeta
		}
	}

	if len(matches) == 1 {
		return matches[0], matchMeta
	}

	// Multiple matches or none: prefer user-scoped secret (no workflow, no org)
	for _, k := range matches {
		meta := SecretNameFromKey(k)
		if meta.WorkflowID == "" && meta.OrgID == "" {
			return k, meta
		}
	}

	return "", nil
}

// A global counter for the task engine
func (n *Engine) NewSeqID() (string, error) {
	num := uint64(0)
	var err error

	defer func() {
		r := recover()
		if r != nil {
			// recover from panic and send err instead
			if e, ok := r.(error); ok {
				err = e
			} else {
				err = fmt.Errorf("panic recovered: %v", r)
			}
		}
	}()

	num, err = n.seq.Next()
	if err != nil { // Check error after first Next() call
		return "", err
	}
	if num == 0 { // This case might indicate an issue with sequence or its initialization if it persists
		n.logger.Warn("Sequence returned 0, attempting Next() again.")
		num, err = n.seq.Next()
		if err != nil {
			return "", err
		}
	}

	return strconv.FormatInt(int64(num), 10), nil
}

func (n *Engine) CanStreamCheck(address string) bool {
	// If no approved operators configured, use default hardcoded list for backward compatibility
	if len(n.config.ApprovedOperators) == 0 {
		n.logger.Debug("Using hardcoded operator approval list", "operator", address)
		return strings.EqualFold(address, "0x997e5d40a32c44a3d93e59fc55c4fd20b7d2d49d") ||
			strings.EqualFold(address, "0xc6b87cc9e85b07365b6abefff061f237f7cf7dc3") ||
			strings.EqualFold(address, "0xa026265a0f01a6e1a19b04655519429df0a57c4e")
	}

	// Check against configured approved operators (case-insensitive)
	for _, approvedAddr := range n.config.ApprovedOperators {
		if strings.EqualFold(address, approvedAddr.Hex()) {
			n.logger.Debug("Operator approved via configuration", "operator", address)
			return true
		}
	}

	n.logger.Debug("Operator not found in approved list", "operator", address, "approved_count", len(n.config.ApprovedOperators))
	return false
}

// shouldLogApprovalMessage checks if we should log the approval message for this operator
// Returns true if more than 3 minutes have passed since the last log for this operator
func (n *Engine) shouldLogApprovalMessage(address string) bool {
	n.approvalLogMutex.Lock()
	defer n.approvalLogMutex.Unlock()

	lastLogTime, exists := n.lastApprovalLogTime[address]
	now := time.Now()

	// Log if no previous log or more than 3 minutes have passed
	if !exists || now.Sub(lastLogTime) >= 3*time.Minute {
		n.lastApprovalLogTime[address] = now
		return true
	}

	return false
}

// supportsTaskTrigger checks if an operator supports a specific trigger type
// supportsTaskChain reports whether an operator's advertised
// supported_chain_ids match a task's chain_id. Pre-multi-chain operators
// (empty list) accept everything for back-compat; tasks with chain_id=0
// are chain-agnostic and accepted by every operator.
func (n *Engine) supportsTaskChain(operatorAddr string, task *model.Workflow) bool {
	n.lock.Lock()
	defer n.lock.Unlock()

	state, exists := n.trackSyncedTasks[operatorAddr]
	if !exists {
		return true
	}
	if len(state.SupportedChainIDs) == 0 {
		return true
	}
	// Capability is judged on the trigger's monitoring chain (G2) — the chain
	// the operator must subscribe to. A chain-agnostic trigger (cron/manual)
	// yields 0 and is covered by any operator.
	monitorChainID := triggerMonitoringChainID(task.Trigger, 0)
	if monitorChainID == 0 {
		return true
	}
	for _, id := range state.SupportedChainIDs {
		if id == monitorChainID {
			return true
		}
	}
	return false
}

func (n *Engine) supportsTaskTrigger(operatorAddr string, task *model.Workflow) bool {
	n.lock.Lock()
	defer n.lock.Unlock()

	// Manual triggers fire via the TriggerWorkflow RPC and execute entirely on
	// the gateway. Operators have no monitoring role for them — neither the
	// operator's monitoring loop nor the NotifyTriggers RPC has any path that
	// handles manual triggers. Reject them before the capability checks so they
	// are never streamed, even to legacy operators that advertise no
	// capabilities.
	if task.Trigger.GetManual() != nil {
		return false
	}

	operatorState, exists := n.trackSyncedTasks[operatorAddr]
	if !exists || operatorState.Capabilities == nil {
		// If no capabilities specified, assume operator supports all trigger types (backward compatibility)
		return true
	}

	capabilities := operatorState.Capabilities

	// Check trigger type support
	if task.Trigger.GetEvent() != nil {
		return capabilities.EventMonitoring
	}
	if task.Trigger.GetBlock() != nil {
		return capabilities.BlockMonitoring
	}
	if task.Trigger.GetCron() != nil || task.Trigger.GetFixedTime() != nil {
		return capabilities.TimeMonitoring
	}

	// Default to true for unknown trigger types
	return true
}

// getEligibleOperators returns operators that support the given task's trigger type
func (n *Engine) getEligibleOperators(task *model.Workflow) []string {
	n.streamsMutex.RLock()
	defer n.streamsMutex.RUnlock()

	var eligible []string
	for operatorAddr := range n.operatorStreams {
		if n.CanStreamCheck(operatorAddr) && n.supportsTaskTrigger(operatorAddr, task) {
			eligible = append(eligible, operatorAddr)
		}
	}

	return eligible
}

// assignTaskToOperator assigns a task to an operator using round-robin
func (n *Engine) assignTaskToOperator(task *model.Workflow) string {
	eligible := n.getEligibleOperators(task)
	if len(eligible) == 0 {
		return ""
	}

	n.assignmentMutex.Lock()
	defer n.assignmentMutex.Unlock()

	// Check if task is already assigned
	if assignedOperator, exists := n.taskAssignments[task.Id]; exists {
		// Verify the assigned operator is still eligible and online
		for _, op := range eligible {
			if op == assignedOperator {
				return assignedOperator
			}
		}
		// Assigned operator is no longer eligible, remove assignment
		delete(n.taskAssignments, task.Id)
	}

	// Round-robin assignment
	selectedOperator := eligible[n.assignmentRoundRobin%len(eligible)]
	n.assignmentRoundRobin++

	// Store assignment
	n.taskAssignments[task.Id] = selectedOperator

	n.logger.Debug("🔄 Round-robin task assignment",
		"task_id", task.Id,
		"assigned_operator", selectedOperator,
		"eligible_operators", len(eligible),
		"round_robin_index", n.assignmentRoundRobin-1)

	return selectedOperator
}

// reassignOrphanedTasks reassigns tasks from disconnected operators
func (n *Engine) reassignOrphanedTasks() {
	// Phase 1: Find and remove orphaned task assignments
	n.assignmentMutex.Lock()

	n.streamsMutex.RLock()
	activeOperators := make(map[string]bool)
	for operatorAddr := range n.operatorStreams {
		activeOperators[operatorAddr] = true
	}
	n.streamsMutex.RUnlock()

	var orphanedTasks []string
	for taskID, operatorAddr := range n.taskAssignments {
		if !activeOperators[operatorAddr] {
			orphanedTasks = append(orphanedTasks, taskID)
			delete(n.taskAssignments, taskID)
		}
	}

	n.assignmentMutex.Unlock()

	if len(orphanedTasks) == 0 {
		return
	}

	n.logger.Info("🔄 Reassigning orphaned tasks",
		"orphaned_count", len(orphanedTasks),
		"active_operators", len(activeOperators))

	if len(activeOperators) == 0 {
		// No active operators to reassign to - leave tasks unassigned for now
		// They will be reclaimed when operators reconnect
		n.logger.Info("⏸️ No active operators available - orphaned tasks will be reclaimed on operator reconnection",
			"orphaned_count", len(orphanedTasks))
		return
	}

	// Phase 2: Snapshot needed tasks under n.lock (lock is the innermost lock, safe to acquire alone)
	n.lock.Lock()
	taskSnapshot := make(map[string]*model.Workflow, len(orphanedTasks))
	for _, taskID := range orphanedTasks {
		if task, exists := n.tasks[taskID]; exists {
			taskSnapshot[taskID] = task
		}
	}
	n.lock.Unlock()

	// Phase 3: Reassign orphaned tasks (assignTaskToOperator handles its own locking)
	reassignmentsByOperator := make(map[string][]string)
	for taskID, task := range taskSnapshot {
		assignedOperator := n.assignTaskToOperator(task)
		if assignedOperator != "" {
			reassignmentsByOperator[assignedOperator] = append(reassignmentsByOperator[assignedOperator], taskID)
		}
	}

	// Log aggregated reassignments per operator
	for operatorAddr, taskIDs := range reassignmentsByOperator {
		n.logger.Info("🔄 Reassigned tasks to operator",
			"operator", operatorAddr,
			"operation", "MonitorTaskTrigger",
			"total_reassignments", len(taskIDs))
	}
}

// GetExecutionStats returns the number of workflows for the given addresses of smart wallets, or if no addresses are provided, it returns the total number of workflows belongs to the requested user
func (n *Engine) GetExecutionStats(user *model.User, payload *avsproto.GetExecutionStatsReq) (*avsproto.GetExecutionStatsResp, error) {
	workflowIds := payload.WorkflowIds
	days := payload.Days
	if days <= 0 {
		days = 7 // Default to 7 days if not specified
	}

	cutoffTime := time.Now().AddDate(0, 0, -int(days)).UnixMilli()

	// Initialize counters
	total := int64(0)
	succeeded := int64(0)
	failed := int64(0)
	var totalExecutionTime int64 = 0

	if len(workflowIds) == 0 {
		workflowIds = []string{}
		var taskIds [][]byte
		var lastErr error
		for _, prefix := range n.chainUserPrefixesBytes(user.Address) {
			chunk, e := n.db.GetKeyHasPrefix(prefix)
			if e != nil {
				lastErr = e
				continue
			}
			taskIds = append(taskIds, chunk...)
		}
		err := lastErr
		_ = err // legacy variable kept; check below operates on err
		if err != nil {
			return nil, status.Errorf(codes.Internal, "Internal error retrieving tasks")
		}
		for _, id := range taskIds {
			taskId := TaskIdFromTaskStatusStorageKey(id)
			workflowIds = append(workflowIds, string(taskId))
		}
	}

	for _, id := range workflowIds {
		if len(id) != TaskIDLength {
			continue
		}

		var items []*storage.KeyValueItem
		for _, prefix := range n.taskExecutionPrefixesBytes(id) {
			chunk, err := n.db.GetByPrefix(prefix)
			if err != nil {
				n.logger.Error("error getting executions", "workflow", id, "error", err)
				continue
			}
			items = append(items, chunk...)
		}

		for _, item := range items {
			execution := &avsproto.Execution{}
			// DiscardUnknown: same rationale as elsewhere — tolerate
			// proto fields renamed/removed since the execution was
			// written.
			if err := (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(item.Value, execution); err != nil {
				n.logger.Error("error unmarshalling execution", "error", err)
				continue
			}

			if execution.StartAt < cutoffTime {
				continue
			}

			total++
			if execution.Status == avsproto.ExecutionStatus_EXECUTION_STATUS_SUCCESS {
				succeeded++
			} else {
				failed++
			}

			if execution.EndAt > execution.StartAt {
				executionTime := execution.EndAt - execution.StartAt
				totalExecutionTime += executionTime
			}
		}
	}

	var avgExecutionTime float64 = 0
	if total > 0 {
		avgExecutionTime = float64(totalExecutionTime) / float64(total)
	}

	return &avsproto.GetExecutionStatsResp{
		Total:            total,
		Succeeded:        succeeded,
		Failed:           failed,
		AvgExecutionTime: avgExecutionTime,
	}, nil
}

func (n *Engine) GetWorkflowCount(user *model.User, payload *avsproto.GetWorkflowCountReq) (*avsproto.GetWorkflowCountResp, error) {
	smartWalletAddresses := payload.Addresses

	total := int64(0)
	var err error

	// Example logic to count workflows
	// This should be replaced with actual logic to count workflows based on addresses
	if len(smartWalletAddresses) == 0 {
		// Default logic: count all tasks for this owner across every chain.
		total, err = n.db.CountKeysByPrefixes(n.chainUserPrefixesBytes(user.Address))
	} else {
		prefixes := [][]byte{}

		for _, address := range smartWalletAddresses {
			smartWalletAddress := common.HexToAddress(address)
			if ok, err := n.userOwnsWalletOnAnyChain(user, smartWalletAddress); !ok || err != nil {
				// skip if the address is not a valid smart wallet address or it isn't belong to this user
				continue
			}

			prefixes = append(prefixes, n.chainSmartWalletPrefixesBytes(user.Address, smartWalletAddress)...)
		}

		total, err = n.db.CountKeysByPrefixes(prefixes)
	}

	if err != nil {
		n.logger.Error("error counting task for", "user", user.Address, "error", err)
		return nil, status.Errorf(codes.Internal, "Internal error counting workflow")
	}

	return &avsproto.GetWorkflowCountResp{
		Total: total,
	}, nil
}

// Helper function to get map keys for logging
func getStringMapKeys(m map[string]interface{}) []string {
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	return keys
}

// GetTokenMetadata handles the RPC for token metadata lookup.
// payload.chain_id is currently informational: in single-chain aggregator mode
// the enrichment service is bound to one chain at startup. Per-chain
// enrichment services are wired through the gateway → worker RPC (Phase 3).
func (n *Engine) GetTokenMetadata(user *model.User, payload *avsproto.GetTokenMetadataReq) (*avsproto.GetTokenMetadataResp, error) {
	// Validate the address parameter
	if payload.Address == "" {
		return &avsproto.GetTokenMetadataResp{
			Found: false,
		}, status.Errorf(codes.InvalidArgument, "token address is required")
	}

	// Check if address is a valid hex address
	if !common.IsHexAddress(payload.Address) {
		return &avsproto.GetTokenMetadataResp{
			Found: false,
		}, status.Errorf(codes.InvalidArgument, "invalid token address format")
	}

	if reqChainID := payload.GetChainId(); reqChainID != 0 && n.logger != nil {
		n.logger.Debug("GetTokenMetadata chain_id requested",
			"address", payload.Address,
			"requested_chain_id", reqChainID)
	}

	// Pick the per-chain TokenEnrichmentService. When the caller specifies a
	// chain, route to that chain's registered service so a Sepolia USDC
	// address is resolved against the Sepolia whitelist/RPC — not whichever
	// chain the engine's default is bound to (chains[0] = mainnet in the
	// production gateway). Falls back to the engine's default when no
	// chain_id is set (single-chain callers / legacy clients).
	tokenService := n.tokenEnrichmentService
	if reqChainID := uint64(payload.GetChainId()); reqChainID != 0 {
		if chainSvc := GetTokenEnrichmentServiceForChain(reqChainID); chainSvc != nil {
			tokenService = chainSvc
		}
	}
	if tokenService == nil {
		return &avsproto.GetTokenMetadataResp{
			Found: false,
		}, status.Errorf(codes.Unavailable, "token enrichment service not available")
	}

	// Try to get token metadata using the enrichment service
	metadata, err := tokenService.GetTokenMetadata(payload.Address)
	if err != nil {
		n.logger.Warn("Failed to get token metadata",
			"address", payload.Address,
			"user", user.Address.Hex(),
			"error", err)

		return &avsproto.GetTokenMetadataResp{
			Found: false,
		}, nil // Return not found instead of error for better UX
	}

	// Check if token was not found (nil metadata but no error)
	if metadata == nil {
		n.logger.Info("Token not found in whitelist or RPC",
			"address", payload.Address,
			"user", user.Address.Hex())

		return &avsproto.GetTokenMetadataResp{
			Found: false,
		}, nil
	}

	// Determine the source of the data
	source := metadata.Source // Use the source from the metadata

	// Return successful response with token metadata
	response := &avsproto.GetTokenMetadataResp{
		Found:  true,
		Source: source,
		Token: &avsproto.TokenMetadata{
			Id:       metadata.Id,
			Name:     metadata.Name,
			Symbol:   metadata.Symbol,
			Decimals: metadata.Decimals,
		},
	}

	n.logger.Info("Token metadata lookup successful",
		"address", payload.Address,
		"user", user.Address.Hex(),
		"tokenName", metadata.Name,
		"tokenSymbol", metadata.Symbol,
		"source", source)

	return response, nil
}

// sanitizeTriggerNameForJS converts trigger names to valid JavaScript variable identifiers
// by replacing spaces and special characters with underscores.
// This ensures trigger names like "my event trigger" or "trigger-1" become valid JS variables
// like "my_event_trigger" and "trigger_1" respectively.
func sanitizeTriggerNameForJS(triggerName string) string {
	if triggerName == "" {
		return "unnamed_trigger"
	}

	// Replace any sequence of non-alphanumeric characters (except underscore) with a single underscore
	reg := regexp.MustCompile(`[^a-zA-Z0-9_]+`)
	sanitized := reg.ReplaceAllString(triggerName, "_")

	// Remove leading/trailing underscores
	sanitized = strings.Trim(sanitized, "_")

	// Ensure it doesn't start with a number (JS variable naming rule)
	if len(sanitized) > 0 && sanitized[0] >= '0' && sanitized[0] <= '9' {
		sanitized = "trigger_" + sanitized
	}

	// Fallback if somehow we end up with empty string
	if sanitized == "" {
		sanitized = "unnamed_trigger"
	}

	return sanitized
}

// buildEventTriggerOutput creates an EventTrigger_Output from raw trigger output data.
// This shared function eliminates code duplication between SimulateTask, RunTriggerRPC,
// and regular task execution flows.
//
// Parameters:
//   - triggerOutput: map containing raw trigger output data from runEventTriggerImmediately
//
// Returns:
//   - *avsproto.EventTrigger_Output: properly structured protobuf output with structured data
func buildEventTriggerOutput(triggerOutput map[string]interface{}) *avsproto.EventTrigger_Output {
	eventOutput := &avsproto.EventTrigger_Output{}

	// Check if we have event data and populate appropriately
	if triggerOutput != nil {
		// Handle both legacy event format (found: true) and new direct calls format (success: true/false)
		var dataToConvert interface{}
		var shouldConvert bool

		// Check for legacy event format first
		if success, ok := triggerOutput["success"].(bool); ok && success {
			// Legacy event format - extract data
			if data, ok := triggerOutput["data"]; ok {
				dataToConvert = data
				shouldConvert = true
			}
		} else if data, hasData := triggerOutput["data"]; hasData {
			// New direct calls format - always include data regardless of success status
			dataToConvert = data
			shouldConvert = true
		}

		// Convert data if we have any
		if shouldConvert && dataToConvert != nil {
			// Handle different data types: JSON string, map, or other types
			var finalData interface{}
			switch d := dataToConvert.(type) {
			case string:
				// Try to parse as JSON string
				var parsedData interface{}
				if err := json.Unmarshal([]byte(d), &parsedData); err == nil {
					finalData = parsedData
				} else {
					// If not valid JSON, treat as plain string (but only if non-empty)
					if d != "" {
						finalData = d
					}
				}
			case map[string]interface{}:
				// Direct map data - always valid
				finalData = d
			default:
				// For defensive programming, only accept complex data types (maps, arrays, strings)
				// Reject simple primitives like numbers, booleans as they're likely invalid event data
				switch d.(type) {
				case []interface{}, []map[string]interface{}:
					// Arrays are valid
					finalData = d
				case int, int32, int64, float32, float64, bool:
					// Simple primitives are considered invalid for event data
					finalData = nil
				default:
					// Other complex types are valid
					finalData = d
				}
			}

			// Convert to google.protobuf.Value only if we have valid data
			if finalData != nil {
				// Convert data to protobuf-compatible format before serialization
				compatibleData := convertToProtobufCompatible(finalData)
				if protoValue, err := structpb.NewValue(compatibleData); err == nil {
					eventOutput.Data = protoValue
				}
			}
		}
	}

	return eventOutput
}

// buildBlockTriggerOutput creates a BlockTrigger_Output from raw trigger output data.
// This shared function eliminates code duplication between SimulateTask, RunTriggerRPC,
// and regular task execution flows.
//
// IMPORTANT: Type Conversion Limitation
// =====================================
// This function converts uint64 values to protobuf Value structures using structpb.NewValue().
// Due to protobuf's internal use of JSON, all numeric types get converted to float64.
// This means:
//   - Input:  blockNumber: uint64(12345)
//   - Output: blockNumber: float64(12345) (via protobuf)
//
// This type conversion happens because:
// 1. structpb.NewValue() uses JSON internally
// 2. JSON only has one numeric type (float64 in Go)
// 3. All integers get converted to float64
//
// Impact:
// - buildTriggerDataMap() preserves uint64 types (works with raw data)
// - buildTriggerDataMapFromProtobuf() returns float64 types (works with protobuf data)
// - This creates inconsistency between different data paths
//
// Client Consistency Requirement:
// The key requirement is that client input should match execution step output.
// As long as users get back the same values they provided, the internal conversion is acceptable.
//
// Potential Solutions:
// 1. Avoid protobuf conversion when not needed (preserve raw data) - requires major refactoring
// 2. Use custom protobuf types that preserve integer types - complex implementation
// 3. Accept the conversion and ensure client consistency - current approach
//
// Currently using solution #3: clients typically send JSON with float64 numbers anyway,
// so the protobuf conversion maintains consistency from the client perspective.
// Tests verify that user input matches execution step output values and types.
//
// Parameters:
//   - triggerOutput: map containing raw trigger output data from runBlockTriggerImmediately
//
// Returns:
//   - *avsproto.BlockTrigger_Output: properly structured protobuf output with block data
//     (note: numeric values will be float64 due to protobuf conversion)
func buildBlockTriggerOutput(triggerOutput map[string]interface{}) *avsproto.BlockTrigger_Output {
	blockNumber := uint64(0)
	blockHash := ""
	timestamp := uint64(0)
	parentHash := ""
	difficulty := ""
	gasLimit := uint64(0)
	gasUsed := uint64(0)

	if triggerOutput != nil {
		if bn, ok := triggerOutput["blockNumber"]; ok {
			if bnUint, ok := bn.(uint64); ok {
				blockNumber = bnUint
			}
		}
		if bh, ok := triggerOutput["blockHash"]; ok {
			if bhStr, ok := bh.(string); ok {
				blockHash = bhStr
			}
		}
		if ts, ok := triggerOutput["timestamp"]; ok {
			if tsUint, ok := ts.(uint64); ok {
				timestamp = tsUint
			}
		}
		if ph, ok := triggerOutput["parentHash"]; ok {
			if phStr, ok := ph.(string); ok {
				parentHash = phStr
			}
		}
		if diff, ok := triggerOutput["difficulty"]; ok {
			if diffStr, ok := diff.(string); ok {
				difficulty = diffStr
			}
		}
		if gl, ok := triggerOutput["gasLimit"]; ok {
			if glUint, ok := gl.(uint64); ok {
				gasLimit = glUint
			}
		}
		if gu, ok := triggerOutput["gasUsed"]; ok {
			if guUint, ok := gu.(uint64); ok {
				gasUsed = guUint
			}
		}
	}

	// Create the data structure with all block information
	blockData := map[string]interface{}{
		"blockNumber": blockNumber,
		"blockHash":   blockHash,
		"timestamp":   timestamp,
		"parentHash":  parentHash,
		"difficulty":  difficulty,
		"gasLimit":    gasLimit,
		"gasUsed":     gasUsed,
	}

	// Convert to protobuf Value
	dataValue, err := structpb.NewValue(blockData)
	if err != nil {
		// Fallback to empty data on error
		dataValue, _ = structpb.NewValue(map[string]interface{}{})
	}

	return &avsproto.BlockTrigger_Output{
		Data: dataValue,
	}
}

// buildFixedTimeTriggerOutput creates a FixedTimeTrigger_Output from raw trigger output data.
// This shared function eliminates code duplication between SimulateTask, RunTriggerRPC,
// and regular task execution flows.
//
// Parameters:
//   - triggerOutput: map containing raw trigger output data from runFixedTimeTriggerImmediately
//
// Returns:
//   - *avsproto.FixedTimeTrigger_Output: properly structured protobuf output with timestamp data
func buildFixedTimeTriggerOutput(triggerOutput map[string]interface{}) *avsproto.FixedTimeTrigger_Output {
	timestamp := uint64(0)
	timestampISO := ""

	if triggerOutput != nil {
		if ts, ok := triggerOutput["timestamp"]; ok {
			if tsUint, ok := ts.(uint64); ok {
				timestamp = tsUint
			}
		}
		if tsISO, ok := triggerOutput["timestamp_iso"]; ok {
			if tsISOStr, ok := tsISO.(string); ok {
				timestampISO = tsISOStr
			}
		}
	}

	// Create the data structure with timestamp information
	timeData := map[string]interface{}{
		"timestamp":    timestamp,
		"timestampIso": timestampISO,
	}

	// Convert to protobuf Value
	dataValue, err := structpb.NewValue(timeData)
	if err != nil {
		// Fallback to empty data on error
		dataValue, _ = structpb.NewValue(map[string]interface{}{})
	}

	return &avsproto.FixedTimeTrigger_Output{
		Data: dataValue,
	}
}

// buildCronTriggerOutput creates a CronTrigger_Output from raw trigger output data.
// This shared function eliminates code duplication between SimulateTask, RunTriggerRPC,
// and regular task execution flows.
//
// Parameters:
//   - triggerOutput: map containing raw trigger output data from runCronTriggerImmediately
//
// Returns:
//   - *avsproto.CronTrigger_Output: properly structured protobuf output with timestamp data
func buildCronTriggerOutput(triggerOutput map[string]interface{}) *avsproto.CronTrigger_Output {
	timestamp := uint64(0)
	timestampISO := ""

	if triggerOutput != nil {
		if ts, ok := triggerOutput["timestamp"]; ok {
			if tsUint, ok := ts.(uint64); ok {
				timestamp = tsUint
			}
		}
		if tsISO, ok := triggerOutput["timestamp_iso"]; ok {
			if tsISOStr, ok := tsISO.(string); ok {
				timestampISO = tsISOStr
			}
		}
	}

	// Create the data structure with timestamp information
	cronData := map[string]interface{}{
		"timestamp":    timestamp,
		"timestampIso": timestampISO,
	}

	// Convert to protobuf Value
	dataValue, err := structpb.NewValue(cronData)
	if err != nil {
		// Fallback to empty data on error
		dataValue, _ = structpb.NewValue(map[string]interface{}{})
	}

	return &avsproto.CronTrigger_Output{
		Data: dataValue,
	}
}

// buildManualTriggerOutput creates a ManualTrigger_Output from raw trigger output data.
// This shared function eliminates code duplication between SimulateTask, RunTriggerRPC,
// and regular task execution flows.
//
// Parameters:
//   - triggerOutput: map containing raw trigger output data from runManualTriggerImmediately
//
// Returns:
//   - *avsproto.ManualTrigger_Output: properly structured protobuf output with only the parsed data
func buildManualTriggerOutput(triggerOutput map[string]interface{}) *avsproto.ManualTrigger_Output {
	var data *structpb.Value

	if triggerOutput != nil {
		// Include ONLY the user-defined JSON data - this is the main payload for manual triggers
		// Headers and pathParams are only used for configuration, not output
		if dataValue, exists := triggerOutput["data"]; exists && dataValue != nil {
			// Convert any valid JSON data (objects, arrays, etc.) to protobuf Value
			if pbValue, err := structpb.NewValue(dataValue); err == nil {
				data = pbValue
			}
		}
	}

	result := &avsproto.ManualTrigger_Output{
		Data: data,
		// Headers and PathParams are removed from output - they're config-only fields
	}
	return result
}

// buildTriggerDataMap creates a map for JavaScript trigger variable access.
// This shared function eliminates code duplication between SimulateTask and VM initialization.
//
// Parameters:
//   - triggerType: the type of trigger being processed
//   - triggerOutput: map containing raw trigger output data
//
// Returns:
//   - map[string]interface{}: JavaScript-accessible trigger data map
func buildTriggerDataMap(triggerType avsproto.TriggerType, triggerOutput map[string]interface{}) map[string]interface{} {
	triggerDataMap := make(map[string]interface{})

	if triggerOutput == nil {
		return triggerDataMap
	}

	switch triggerType {
	case avsproto.TriggerType_TRIGGER_TYPE_MANUAL:
		// For manual triggers, include all fields (data, headers, pathParams) for template access
		// This allows templates to access ManualTrigger.data.field, ManualTrigger.headers.field, etc.
		for k, v := range triggerOutput {
			triggerDataMap[k] = v
		}
	case avsproto.TriggerType_TRIGGER_TYPE_FIXED_TIME:
		if timestamp, ok := triggerOutput["timestamp"]; ok {
			triggerDataMap["timestamp"] = timestamp
		}
		if timestampISO, ok := triggerOutput["timestamp_iso"]; ok {
			triggerDataMap["timestamp_iso"] = timestampISO
		}
	case avsproto.TriggerType_TRIGGER_TYPE_CRON:
		if timestamp, ok := triggerOutput["timestamp"]; ok {
			triggerDataMap["timestamp"] = timestamp
		}
		if timestampISO, ok := triggerOutput["timestamp_iso"]; ok {
			triggerDataMap["timestamp_iso"] = timestampISO
		}
	case avsproto.TriggerType_TRIGGER_TYPE_BLOCK:
		if blockNumber, ok := triggerOutput["blockNumber"]; ok {
			triggerDataMap["blockNumber"] = blockNumber
		}
		if blockHash, ok := triggerOutput["blockHash"]; ok {
			triggerDataMap["blockHash"] = blockHash
		}
		if timestamp, ok := triggerOutput["timestamp"]; ok {
			triggerDataMap["timestamp"] = timestamp
		}
		if parentHash, ok := triggerOutput["parentHash"]; ok {
			triggerDataMap["parentHash"] = parentHash
		}
		if difficulty, ok := triggerOutput["difficulty"]; ok {
			triggerDataMap["difficulty"] = difficulty
		}
		if gasLimit, ok := triggerOutput["gasLimit"]; ok {
			triggerDataMap["gasLimit"] = gasLimit
		}
		if gasUsed, ok := triggerOutput["gasUsed"]; ok {
			triggerDataMap["gasUsed"] = gasUsed
		}
	case avsproto.TriggerType_TRIGGER_TYPE_EVENT:
		// Handle event trigger data with special processing for transfer_log
		if transferLogData, hasTransferLog := triggerOutput["transfer_log"].(map[string]interface{}); hasTransferLog {
			// Flatten transfer_log data to top level for JavaScript access
			for k, v := range transferLogData {
				triggerDataMap[k] = v
			}
		} else {
			// Check for enhanced response format first (new format with success field)
			if successValue, hasSuccess := triggerOutput["success"]; hasSuccess {
				// Enhanced response format - extract data appropriately
				if successBool, ok := successValue.(bool); ok && !successBool {
					// Conditions not met - data is in error message as JSON string
					if errorValue, hasError := triggerOutput["error"]; hasError {
						if errorStr, ok := errorValue.(string); ok {
							// Try to extract data from error message
							if strings.HasPrefix(errorStr, "Conditions not met. Data: ") {
								jsonStr := strings.TrimPrefix(errorStr, "Conditions not met. Data: ")
								var errorData map[string]interface{}
								if err := json.Unmarshal([]byte(jsonStr), &errorData); err == nil {
									// Successfully extracted data from error message
									for k, v := range errorData {
										if k != "conditions" { // Skip conditions metadata
											triggerDataMap[k] = v
										}
									}
								}
							}
						}
					}
				} else {
					// Conditions met - data is in data field
					if eventData, hasEventData := triggerOutput["data"].(map[string]interface{}); hasEventData {
						for k, v := range eventData {
							triggerDataMap[k] = v
						}
					} else {
						// No data field but success=true - copy all fields as-is
						// This handles cases where event data is directly in the triggerOutput
						for k, v := range triggerOutput {
							triggerDataMap[k] = v
						}
					}
				}
			} else {
				// Legacy format - check if this is a simulation result structure
				// Simulation results should have "success", "metadata", and "data" fields
				if _, hasSuccess := triggerOutput["success"]; hasSuccess {
					if _, hasMetadata := triggerOutput["metadata"]; hasMetadata {
						if eventData, hasEventData := triggerOutput["data"].(map[string]interface{}); hasEventData {
							// Extract the actual event data from the nested "data" field
							for k, v := range eventData {
								triggerDataMap[k] = v
							}
						} else {
							// No valid data field in simulation result - copy all data as-is
							for k, v := range triggerOutput {
								triggerDataMap[k] = v
							}
						}
					} else {
						// Not a complete simulation result structure - copy all data as-is
						for k, v := range triggerOutput {
							triggerDataMap[k] = v
						}
					}
				} else {
					// Not a simulation result structure - this should be actual event data
					// Copy all event trigger data directly
					for k, v := range triggerOutput {
						triggerDataMap[k] = v
					}
				}
			}
		}
	default:
		// For unknown trigger types, copy all output data
		for k, v := range triggerOutput {
			triggerDataMap[k] = v
		}
	}

	return triggerDataMap
}

// Shared functions for eventTrigger response processing
// These functions are used by both runNodeImmediately and simulateWorkflow

// extractTriggerStatus extracts success/error status from trigger output
func extractTriggerStatus(triggerOutput map[string]interface{}) (bool, string) {
	if triggerOutput == nil {
		return true, ""
	}

	// Check for enhanced response format with success field
	if successValue, hasSuccess := triggerOutput["success"]; hasSuccess {
		if successBool, ok := successValue.(bool); ok {
			if !successBool {
				// Extract error message for failed triggers
				if errorValue, hasError := triggerOutput["error"]; hasError {
					if errorStr, ok := errorValue.(string); ok {
						return false, errorStr
					}
				}
				return false, "Conditions not met"
			}
			return true, ""
		}
	}

	// Legacy format - assume success if no explicit failure
	return true, ""
}

// extractTriggerMetadata extracts metadata from trigger output for runNodeImmediately
func extractTriggerMetadata(triggerOutput map[string]interface{}) interface{} {
	if triggerOutput == nil {
		return nil
	}

	// Check for metadata field (can be array or map)
	if metadataValue, hasMetadata := triggerOutput["metadata"]; hasMetadata {
		// New format - array of method responses (like contract_read)
		if metadataArray, ok := metadataValue.([]interface{}); ok {
			return metadataArray
		}
		// Legacy format - map
		if metadataMap, ok := metadataValue.(map[string]interface{}); ok {
			return metadataMap
		}
	}

	return nil
}

// extractTriggerExecutionContext extracts execution context from trigger output
func extractTriggerExecutionContext(triggerOutput map[string]interface{}) map[string]interface{} {
	if triggerOutput == nil {
		return nil
	}

	if execCtxValue, hasExecCtx := triggerOutput["executionContext"]; hasExecCtx {
		if execCtxMap, ok := execCtxValue.(map[string]interface{}); ok {
			return execCtxMap
		}
	}

	return nil
}

// buildExecutionStepOutputData creates the appropriate OutputData oneof field for execution steps.
// This shared function eliminates code duplication in execution step creation.
//
// Parameters:
//   - triggerType: the type of trigger being processed
//   - triggerOutputProto: the protobuf trigger output structure
//
// Returns:
//   - avsproto.IsExecution_Step_OutputData: the appropriate oneof field for the execution step
func buildExecutionStepOutputData(triggerType avsproto.TriggerType, triggerOutputProto interface{}) avsproto.IsExecution_Step_OutputData {
	if triggerOutputProto == nil {
		// Create empty output structure based on trigger type to avoid nil output
		switch triggerType {
		case avsproto.TriggerType_TRIGGER_TYPE_MANUAL:
			return &avsproto.Execution_Step_ManualTrigger{ManualTrigger: &avsproto.ManualTrigger_Output{}}
		case avsproto.TriggerType_TRIGGER_TYPE_FIXED_TIME:
			return &avsproto.Execution_Step_FixedTimeTrigger{FixedTimeTrigger: &avsproto.FixedTimeTrigger_Output{}}
		case avsproto.TriggerType_TRIGGER_TYPE_CRON:
			return &avsproto.Execution_Step_CronTrigger{CronTrigger: &avsproto.CronTrigger_Output{}}
		case avsproto.TriggerType_TRIGGER_TYPE_BLOCK:
			return &avsproto.Execution_Step_BlockTrigger{BlockTrigger: &avsproto.BlockTrigger_Output{}}
		case avsproto.TriggerType_TRIGGER_TYPE_EVENT:
			return &avsproto.Execution_Step_EventTrigger{EventTrigger: &avsproto.EventTrigger_Output{
				// Empty EventTrigger output with no data
			}}
		}
		return nil
	}

	switch triggerType {
	case avsproto.TriggerType_TRIGGER_TYPE_MANUAL:
		if output, ok := triggerOutputProto.(*avsproto.ManualTrigger_Output); ok {
			return &avsproto.Execution_Step_ManualTrigger{ManualTrigger: output}
		}
	case avsproto.TriggerType_TRIGGER_TYPE_FIXED_TIME:
		if output, ok := triggerOutputProto.(*avsproto.FixedTimeTrigger_Output); ok {
			return &avsproto.Execution_Step_FixedTimeTrigger{FixedTimeTrigger: output}
		}
	case avsproto.TriggerType_TRIGGER_TYPE_CRON:
		if output, ok := triggerOutputProto.(*avsproto.CronTrigger_Output); ok {
			return &avsproto.Execution_Step_CronTrigger{CronTrigger: output}
		}
	case avsproto.TriggerType_TRIGGER_TYPE_BLOCK:
		if output, ok := triggerOutputProto.(*avsproto.BlockTrigger_Output); ok {
			return &avsproto.Execution_Step_BlockTrigger{BlockTrigger: output}
		}
	case avsproto.TriggerType_TRIGGER_TYPE_EVENT:
		if output, ok := triggerOutputProto.(*avsproto.EventTrigger_Output); ok {
			return &avsproto.Execution_Step_EventTrigger{EventTrigger: output}
		}
		// If type assertion failed, log the actual type for debugging
		if triggerOutputProto != nil {
			// Create empty EventTrigger output as fallback to avoid nil
			return &avsproto.Execution_Step_EventTrigger{EventTrigger: &avsproto.EventTrigger_Output{
				// Empty EventTrigger output with no data
			}}
		}
	}

	// Fallback: create empty output structure based on trigger type to avoid nil output
	switch triggerType {
	case avsproto.TriggerType_TRIGGER_TYPE_MANUAL:
		return &avsproto.Execution_Step_ManualTrigger{ManualTrigger: &avsproto.ManualTrigger_Output{}}
	case avsproto.TriggerType_TRIGGER_TYPE_FIXED_TIME:
		return &avsproto.Execution_Step_FixedTimeTrigger{FixedTimeTrigger: &avsproto.FixedTimeTrigger_Output{}}
	case avsproto.TriggerType_TRIGGER_TYPE_CRON:
		return &avsproto.Execution_Step_CronTrigger{CronTrigger: &avsproto.CronTrigger_Output{}}
	case avsproto.TriggerType_TRIGGER_TYPE_BLOCK:
		return &avsproto.Execution_Step_BlockTrigger{BlockTrigger: &avsproto.BlockTrigger_Output{}}
	case avsproto.TriggerType_TRIGGER_TYPE_EVENT:
		return &avsproto.Execution_Step_EventTrigger{EventTrigger: &avsproto.EventTrigger_Output{
			// Empty EventTrigger output with no data
		}}
	}

	return nil
}

// buildTriggerDataMapFromProtobuf creates a map for JavaScript trigger variable access from protobuf structures.
// This shared function eliminates code duplication in VM initialization where protobuf trigger outputs
// need to be converted to JavaScript-accessible data maps.
//
// Parameters:
//   - triggerType: the type of trigger being processed
//   - triggerOutputProto: the protobuf trigger output structure (e.g., *avsproto.BlockTrigger_Output)
//   - logger: optional logger for debugging (can be nil)
//
// Returns:
//   - map[string]interface{}: JavaScript-accessible trigger data map
//
// This function differs from buildTriggerDataMap in that it works with structured protobuf data
// rather than raw trigger output maps, making it suitable for VM initialization.
func buildTriggerDataMapFromProtobuf(triggerType avsproto.TriggerType, triggerOutputProto interface{}, logger sdklogging.Logger) map[string]interface{} {
	triggerDataMap := make(map[string]interface{})

	if triggerOutputProto == nil {
		// Always add trigger type for reference if available
		if triggerType != avsproto.TriggerType_TRIGGER_TYPE_UNSPECIFIED {
			triggerDataMap["type"] = triggerType.String()
		}
		return triggerDataMap
	}

	switch triggerType {
	case avsproto.TriggerType_TRIGGER_TYPE_MANUAL:
		if manualOutput, ok := triggerOutputProto.(*avsproto.ManualTrigger_Output); ok && manualOutput.Data != nil {
			triggerDataMap["data"] = manualOutput.Data.AsInterface()
		} else if mapOutput, ok := triggerOutputProto.(map[string]interface{}); ok {
			if dataField, exists := mapOutput["data"]; exists {
				triggerDataMap["data"] = dataField
			}
		}
	case avsproto.TriggerType_TRIGGER_TYPE_FIXED_TIME:
		if timeOutput, ok := triggerOutputProto.(*avsproto.FixedTimeTrigger_Output); ok {
			// Extract data from the new standardized data field
			if timeOutput.Data != nil {
				dataMap := gow.ValueToMap(timeOutput.Data)
				if dataMap != nil {
					if timestamp, ok := dataMap["timestamp"]; ok {
						triggerDataMap["timestamp"] = timestamp
					}
					if timestampIso, ok := dataMap["timestampIso"]; ok {
						triggerDataMap["timestamp_iso"] = timestampIso
					}
				}
			}
		}
	case avsproto.TriggerType_TRIGGER_TYPE_CRON:
		if cronOutput, ok := triggerOutputProto.(*avsproto.CronTrigger_Output); ok {
			// Extract data from the new standardized data field
			if cronOutput.Data != nil {
				dataMap := gow.ValueToMap(cronOutput.Data)
				if dataMap != nil {
					if timestamp, ok := dataMap["timestamp"]; ok {
						triggerDataMap["timestamp"] = timestamp
					}
					if timestampIso, ok := dataMap["timestampIso"]; ok {
						triggerDataMap["timestamp_iso"] = timestampIso
					}
				}
			}
		}
	case avsproto.TriggerType_TRIGGER_TYPE_BLOCK:
		if blockOutput, ok := triggerOutputProto.(*avsproto.BlockTrigger_Output); ok {
			// Extract data from the new standardized data field
			if blockOutput.Data != nil {
				dataMap := gow.ValueToMap(blockOutput.Data)
				if dataMap != nil {
					if blockNumber, ok := dataMap["blockNumber"]; ok {
						triggerDataMap["blockNumber"] = blockNumber
					}
					if blockHash, ok := dataMap["blockHash"]; ok {
						triggerDataMap["blockHash"] = blockHash
					}
					if timestamp, ok := dataMap["timestamp"]; ok {
						triggerDataMap["timestamp"] = timestamp
					}
					if parentHash, ok := dataMap["parentHash"]; ok {
						triggerDataMap["parentHash"] = parentHash
					}
					if difficulty, ok := dataMap["difficulty"]; ok {
						triggerDataMap["difficulty"] = difficulty
					}
					if gasLimit, ok := dataMap["gasLimit"]; ok {
						triggerDataMap["gasLimit"] = gasLimit
					}
					if gasUsed, ok := dataMap["gasUsed"]; ok {
						triggerDataMap["gasUsed"] = gasUsed
					}
				}
			}
		} else if rawMap, ok := triggerOutputProto.(map[string]interface{}); ok {
			// Handle raw map data (from queue/storage) - convert snake_case to camelCase
			if blockNumber, exists := rawMap["block_number"]; exists {
				triggerDataMap["blockNumber"] = blockNumber
			}
			if blockHash, exists := rawMap["block_hash"]; exists {
				triggerDataMap["blockHash"] = blockHash
			}
			if timestamp, exists := rawMap["timestamp"]; exists {
				triggerDataMap["timestamp"] = timestamp
			}
			if parentHash, exists := rawMap["parent_hash"]; exists {
				triggerDataMap["parentHash"] = parentHash
			}
			if difficulty, exists := rawMap["difficulty"]; exists {
				triggerDataMap["difficulty"] = difficulty
			}
			if gasLimit, exists := rawMap["gas_limit"]; exists {
				triggerDataMap["gasLimit"] = gasLimit
			}
			if gasUsed, exists := rawMap["gas_used"]; exists {
				triggerDataMap["gasUsed"] = gasUsed
			}
		}
	case avsproto.TriggerType_TRIGGER_TYPE_EVENT:
		if eventOutput, ok := triggerOutputProto.(*avsproto.EventTrigger_Output); ok {
			// With new structured data, convert the protobuf value to map
			if eventOutput.Data != nil {
				// Convert google.protobuf.Value to map[string]interface{}
				if eventData, ok := eventOutput.Data.AsInterface().(map[string]interface{}); ok {
					// Copy all parsed event data to the trigger data map
					for k, v := range eventData {
						triggerDataMap[k] = v
					}
				} else if logger != nil {
					logger.Warn("Failed to convert event trigger data to map", "data_type", fmt.Sprintf("%T", eventOutput.Data.AsInterface()))
				}
			}
		} else if enrichedDataMap, ok := triggerOutputProto.(map[string]interface{}); ok {
			// Handle the new enriched data format that survives JSON serialization
			if enrichedData, hasEnrichedData := enrichedDataMap["enriched_data"].(map[string]interface{}); hasEnrichedData {
				// Copy all enriched data to the trigger data map
				for k, v := range enrichedData {
					triggerDataMap[k] = v
				}
				if logger != nil {
					logger.Debug("loaded enriched event trigger data from queue",
						"has_token_symbol", triggerDataMap["tokenSymbol"] != nil,
						"has_value_formatted", triggerDataMap["valueFormatted"] != nil)
				}
			} else {
				// Fallback: copy all data from the map
				for k, v := range enrichedDataMap {
					if k != "trigger_type" { // Skip metadata
						triggerDataMap[k] = v
					}
				}
			}
		}
	default:
		// For unknown trigger types, return empty map
	}

	// Always add trigger type for reference
	if triggerType != avsproto.TriggerType_TRIGGER_TYPE_UNSPECIFIED {
		triggerDataMap["type"] = triggerType.String()
	}

	return triggerDataMap
}

// DetectAndHandleInvalidTasks scans for tasks with invalid configurations
// and either marks them as failed or removes them based on the strategy
func (n *Engine) DetectAndHandleInvalidTasks() error {
	n.logger.Info("🔍 Scanning for tasks with invalid configurations...")

	invalidTasks := []string{}
	updates := make(map[string][]byte)
	// Status-keyed storage uses status in the key (t:<chain>:<status>:<id>),
	// so flipping a task to Failed means writing the new Failed key AND
	// deleting the old (Enabled) one. Without the delete, the stale Enabled
	// key survives, MustStart reloads it from the Enabled prefix on the next
	// boot, and this scan re-fails it — an invisible every-boot loop. Collect
	// the stale keys here and delete them after the batch write succeeds.
	staleStatusKeys := [][]byte{}

	// Acquire lock once for the entire operation to reduce lock contention
	n.lock.Lock()

	// Scan through all tasks in memory and prepare updates
	for taskID, task := range n.tasks {
		if err := task.ValidateWithError(); err != nil {
			invalidTasks = append(invalidTasks, taskID)

			// Log owner/smart-wallet/created-at so an auto-failed task can be
			// traced to its user straight from the logs, without a DB query —
			// the observability gap that let a cohort of invalid tasks fail
			// silently for ~a year. created_at is derived from the ULID task
			// ID since Task has no created_at field.
			createdAt := "unknown"
			if parsed, perr := ulid.Parse(strings.ToUpper(taskID)); perr == nil {
				createdAt = time.UnixMilli(int64(parsed.Time())).UTC().Format(time.RFC3339)
			}
			n.logger.Warn("🚨 Found invalid task configuration",
				"task_id", taskID,
				"owner", task.GetOwner(),
				"smart_wallet", task.GetSmartWalletAddress(),
				"created_at", createdAt,
				"error", err.Error())

			// Capture the pre-failure status key before SetFailed mutates it.
			oldStatus := task.Status

			// Mark task as failed and prepare storage updates
			task.SetFailed()

			taskJSON, err := task.ToJSON()
			if err != nil {
				n.logger.Error("Failed to serialize invalid task for cleanup",
					"task_id", taskID,
					"error", err)
				continue
			}

			// Prepare the task status update in storage
			updates[string(WorkflowStorageKey(task.Id, task.Status))] = taskJSON
			updates[string(TaskUserKey(task))] = []byte(fmt.Sprintf("%d", avsproto.TaskStatus_Failed))

			// Queue the old status-keyed row for deletion (skip when the
			// status didn't actually change, to avoid deleting what we just
			// wrote).
			if oldStatus != task.Status {
				staleStatusKeys = append(staleStatusKeys, WorkflowStorageKey(task.Id, oldStatus))
			}

			// Drop from the in-memory active set so it isn't treated as active
			// for the rest of this process run. Deleting the current key while
			// ranging over a map is safe in Go.
			delete(n.tasks, taskID)
		}
	}

	n.lock.Unlock()

	if len(invalidTasks) == 0 {
		n.logger.Info("✅ No invalid tasks found")
		return nil
	}

	n.logger.Warn("🚨 Found invalid tasks, marking as failed",
		"count", len(invalidTasks),
		"task_ids", invalidTasks)

	// Write the Failed records first so we never lose a task record, ...
	if len(updates) > 0 {
		if err := n.db.BatchWrite(updates); err != nil {
			n.logger.Error("Failed to update invalid tasks in storage",
				"error", err)
			return err
		}
	}

	// ... then delete the stale old-status rows so the next boot's
	// Enabled-prefix scan doesn't reload and re-fail them.
	for _, key := range staleStatusKeys {
		if err := n.db.Delete(key); err != nil {
			n.logger.Error("Failed to delete stale task status key",
				"key", string(key),
				"error", err)
		}
	}

	n.logger.Info("✅ Successfully marked invalid tasks as failed",
		"count", len(invalidTasks))

	return nil
}