Skip to main content

arete_interpreter/
vm.rs

1use crate::ast::{
2    self, BinaryOp, ComparisonOp, ComputedExpr, ComputedFieldSpec, FieldPath, ResolveStrategy,
3    ResolverExtractSpec, ResolverType, Transformation,
4};
5use crate::compiler::{MultiEntityBytecode, OpCode};
6use crate::debugger::{VmDebugEvent, VmDebugger, VmLookupHop};
7use crate::Mutation;
8use dashmap::DashMap;
9use lru::LruCache;
10use once_cell::sync::Lazy;
11use serde_json::{json, Value};
12use std::collections::{HashMap, HashSet, VecDeque};
13use std::num::NonZeroUsize;
14use std::sync::Arc;
15use std::time::{Duration, Instant};
16
17#[cfg(feature = "otel")]
18use tracing::instrument;
19/// Context metadata for blockchain updates (accounts and instructions)
20/// This structure is designed to be extended over time with additional metadata
21#[derive(Debug, Clone, Default)]
22pub struct UpdateContext {
23    /// Blockchain slot number
24    pub slot: Option<u64>,
25    /// Transaction signature
26    pub signature: Option<String>,
27    /// Unix timestamp (seconds since epoch)
28    /// If not provided, will default to current system time when accessed
29    pub timestamp: Option<i64>,
30    /// Write version for account updates (monotonically increasing per account within a slot)
31    /// Used for staleness detection to reject out-of-order updates
32    pub write_version: Option<u64>,
33    /// Transaction index for instruction updates (orders transactions within a slot)
34    /// Used for staleness detection to reject out-of-order updates
35    pub txn_index: Option<u64>,
36    /// When true, immediate resolver execution is skipped during handler execution.
37    /// Future scheduled callbacks are still registered after a PDA remap so fresh
38    /// account data can resolve at its intended slot.
39    pub skip_resolvers: bool,
40    /// Additional custom metadata that can be added without breaking changes
41    pub metadata: HashMap<String, Value>,
42}
43
44impl UpdateContext {
45    /// Create a new UpdateContext with slot and signature
46    pub fn new(slot: u64, signature: String) -> Self {
47        Self {
48            slot: Some(slot),
49            signature: Some(signature),
50            timestamp: None,
51            write_version: None,
52            txn_index: None,
53            skip_resolvers: false,
54            metadata: HashMap::new(),
55        }
56    }
57
58    /// Create a new UpdateContext with slot, signature, and timestamp
59    pub fn with_timestamp(slot: u64, signature: String, timestamp: i64) -> Self {
60        Self {
61            slot: Some(slot),
62            signature: Some(signature),
63            timestamp: Some(timestamp),
64            write_version: None,
65            txn_index: None,
66            skip_resolvers: false,
67            metadata: HashMap::new(),
68        }
69    }
70
71    /// Create context for account updates with write_version for staleness detection
72    pub fn new_account(slot: u64, signature: String, write_version: u64) -> Self {
73        Self {
74            slot: Some(slot),
75            signature: Some(signature),
76            timestamp: None,
77            write_version: Some(write_version),
78            txn_index: None,
79            skip_resolvers: false,
80            metadata: HashMap::new(),
81        }
82    }
83
84    /// Create context for instruction updates with txn_index for staleness detection
85    pub fn new_instruction(slot: u64, signature: String, txn_index: u64) -> Self {
86        Self {
87            slot: Some(slot),
88            signature: Some(signature),
89            timestamp: None,
90            write_version: None,
91            txn_index: Some(txn_index),
92            skip_resolvers: false,
93            metadata: HashMap::new(),
94        }
95    }
96
97    /// Create context for reprocessed cached account data from PDA mapping changes.
98    /// Uses an empty signature to prevent `when` guards from matching stale
99    /// instructions and blocks immediate resolver execution to avoid SetOnce lock-in.
100    pub fn new_reprocessed(slot: u64, write_version: u64) -> Self {
101        Self {
102            slot: Some(slot),
103            signature: None,
104            timestamp: None,
105            write_version: Some(write_version),
106            txn_index: None,
107            skip_resolvers: true,
108            metadata: HashMap::new(),
109        }
110    }
111
112    /// Get the timestamp, falling back to current system time if not set
113    pub fn timestamp(&self) -> i64 {
114        self.timestamp.unwrap_or_else(|| {
115            std::time::SystemTime::now()
116                .duration_since(std::time::UNIX_EPOCH)
117                .unwrap()
118                .as_secs() as i64
119        })
120    }
121
122    /// Create an empty context (for testing or when context is not available)
123    pub fn empty() -> Self {
124        Self::default()
125    }
126
127    /// Add custom metadata
128    /// Returns true if this is an account update context (has write_version, no txn_index)
129    pub fn is_account_update(&self) -> bool {
130        self.write_version.is_some() && self.txn_index.is_none()
131    }
132
133    /// Returns true if this is an instruction update context (has txn_index, no write_version)
134    pub fn is_instruction_update(&self) -> bool {
135        self.txn_index.is_some() && self.write_version.is_none()
136    }
137
138    pub fn with_metadata(mut self, key: String, value: Value) -> Self {
139        self.metadata.insert(key, value);
140        self
141    }
142
143    /// Get metadata value
144    pub fn get_metadata(&self, key: &str) -> Option<&Value> {
145        self.metadata.get(key)
146    }
147
148    /// Convert context to JSON value for injection into event data
149    pub fn to_value(&self) -> Value {
150        let mut obj = serde_json::Map::new();
151        if let Some(slot) = self.slot {
152            obj.insert("slot".to_string(), json!(slot));
153        }
154        if let Some(ref sig) = self.signature {
155            obj.insert("signature".to_string(), json!(sig));
156        }
157        // Always include timestamp (use current time if not set)
158        obj.insert("timestamp".to_string(), json!(self.timestamp()));
159        for (key, value) in &self.metadata {
160            obj.insert(key.clone(), value.clone());
161        }
162        Value::Object(obj)
163    }
164}
165
166pub type Register = usize;
167pub type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;
168pub type ComputedEvaluatorResult =
169    std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>;
170
171pub type RegisterValue = Value;
172
173/// Trait for evaluating computed fields
174/// Implement this in your generated spec to enable computed field evaluation
175pub trait ComputedFieldsEvaluator {
176    fn evaluate(&self, state: &mut Value) -> Result<()>;
177}
178
179// Pending queue configuration
180const MAX_PENDING_UPDATES_TOTAL: usize = 2_500;
181const MAX_PENDING_UPDATES_PER_PDA: usize = 50;
182const PENDING_UPDATE_TTL_SECONDS: i64 = 300; // 5 minutes
183
184// Temporal index configuration - prevents unbounded history growth
185const TEMPORAL_HISTORY_TTL_SECONDS: i64 = 300; // 5 minutes, matches pending queue TTL
186const MAX_TEMPORAL_ENTRIES_PER_KEY: usize = 250;
187
188// State table configuration - aligned with downstream EntityCache (500 per view)
189const DEFAULT_MAX_STATE_TABLE_ENTRIES: usize = 2_500;
190const DEFAULT_MAX_ARRAY_LENGTH: usize = 100;
191
192const DEFAULT_MAX_LOOKUP_INDEX_ENTRIES: usize = 2_500;
193
194const DEFAULT_MAX_VERSION_TRACKER_ENTRIES: usize = 2_500;
195
196// Smaller cache for instruction deduplication - provides shorter effective TTL
197// since we don't expect instruction duplicates to arrive much later
198const DEFAULT_MAX_INSTRUCTION_DEDUP_ENTRIES: usize = 500;
199
200const DEFAULT_MAX_TEMPORAL_INDEX_KEYS: usize = 2_500;
201
202const DEFAULT_MAX_PDA_REVERSE_LOOKUP_ENTRIES: usize = 2_500;
203
204const DEFAULT_MAX_RESOLVER_CACHE_ENTRIES: usize = 20_000;
205const DEFAULT_RESOLVER_CACHE_TTL_SECS: u64 = 3600; // 1 hour
206const DEFAULT_NEGATIVE_RESOLVER_CACHE_TTL_SECS: u64 = 30;
207const DEFAULT_RESOLVER_RETRY_BACKOFF_SECS: u64 = 2;
208const DEFAULT_RESOLVER_MAX_RETRIES: u32 = 1;
209
210static RESOLVER_CACHE_CAPACITY: Lazy<NonZeroUsize> = Lazy::new(|| {
211    NonZeroUsize::new(
212        std::env::var("ARETE_RESOLVER_CACHE_CAPACITY")
213            .ok()
214            .and_then(|value| value.parse::<usize>().ok())
215            .filter(|value| *value > 0)
216            .unwrap_or(DEFAULT_MAX_RESOLVER_CACHE_ENTRIES),
217    )
218    .expect("resolver cache capacity must be > 0")
219});
220
221static RESOLVER_CACHE_TTL: Lazy<Duration> = Lazy::new(|| {
222    let ttl_secs = match std::env::var("ARETE_RESOLVER_CACHE_TTL_SECS") {
223        Ok(value) => match value.parse::<u64>() {
224            Ok(0) => {
225                tracing::warn!(
226                    default_ttl_secs = DEFAULT_RESOLVER_CACHE_TTL_SECS,
227                    "ARETE_RESOLVER_CACHE_TTL_SECS=0 is not supported; using default"
228                );
229                DEFAULT_RESOLVER_CACHE_TTL_SECS
230            }
231            Ok(value) => value,
232            Err(_) => DEFAULT_RESOLVER_CACHE_TTL_SECS,
233        },
234        Err(_) => DEFAULT_RESOLVER_CACHE_TTL_SECS,
235    };
236
237    Duration::from_secs(ttl_secs)
238});
239
240static NEGATIVE_RESOLVER_CACHE_TTL: Lazy<Duration> = Lazy::new(|| {
241    let ttl_secs = match std::env::var("ARETE_NEGATIVE_RESOLVER_CACHE_TTL_SECS") {
242        Ok(value) => match value.parse::<u64>() {
243            Ok(0) => {
244                tracing::warn!(
245                    default_ttl_secs = DEFAULT_NEGATIVE_RESOLVER_CACHE_TTL_SECS,
246                    "ARETE_NEGATIVE_RESOLVER_CACHE_TTL_SECS=0 is not supported; using default"
247                );
248                DEFAULT_NEGATIVE_RESOLVER_CACHE_TTL_SECS
249            }
250            Ok(value) => value,
251            Err(_) => DEFAULT_NEGATIVE_RESOLVER_CACHE_TTL_SECS,
252        },
253        Err(_) => DEFAULT_NEGATIVE_RESOLVER_CACHE_TTL_SECS,
254    };
255
256    Duration::from_secs(ttl_secs)
257});
258
259static RESOLVER_RETRY_BACKOFF: Lazy<Duration> = Lazy::new(|| {
260    let secs = std::env::var("ARETE_RESOLVER_RETRY_BACKOFF_SECS")
261        .ok()
262        .and_then(|value| value.parse::<u64>().ok())
263        .filter(|value| *value > 0)
264        .unwrap_or(DEFAULT_RESOLVER_RETRY_BACKOFF_SECS);
265    Duration::from_secs(secs)
266});
267
268static RESOLVER_MAX_RETRIES: Lazy<u32> = Lazy::new(|| {
269    std::env::var("ARETE_RESOLVER_MAX_RETRIES")
270        .ok()
271        .and_then(|value| value.parse::<u32>().ok())
272        .unwrap_or(DEFAULT_RESOLVER_MAX_RETRIES)
273});
274
275#[derive(Debug, Clone)]
276pub(crate) enum CachedResolverValue {
277    Resolved(Value),
278    Negative,
279}
280
281#[derive(Debug, Clone)]
282enum ResolverCacheValue {
283    Resolved(Value),
284    Negative,
285}
286
287#[derive(Debug, Clone)]
288struct ResolverCacheEntry {
289    value: ResolverCacheValue,
290    cached_at: Instant,
291    ttl: Duration,
292}
293
294fn resolver_cache_capacity() -> NonZeroUsize {
295    *RESOLVER_CACHE_CAPACITY
296}
297
298fn resolver_cache_ttl() -> Duration {
299    *RESOLVER_CACHE_TTL
300}
301
302fn negative_resolver_cache_ttl() -> Duration {
303    *NEGATIVE_RESOLVER_CACHE_TTL
304}
305
306fn resolver_retry_backoff() -> Duration {
307    *RESOLVER_RETRY_BACKOFF
308}
309
310fn resolver_max_retries() -> u32 {
311    *RESOLVER_MAX_RETRIES
312}
313
314#[derive(Debug, Clone, Copy, PartialEq, Eq)]
315pub(crate) enum TokenResolverMissAction {
316    RetryScheduled,
317    NegativeCached,
318    Dropped,
319}
320
321/// Estimate the size of a JSON value in bytes
322fn estimate_json_size(value: &Value) -> usize {
323    match value {
324        Value::Null => 4,
325        Value::Bool(_) => 5,
326        Value::Number(_) => 8,
327        Value::String(s) => s.len() + 2,
328        Value::Array(arr) => 2 + arr.iter().map(|v| estimate_json_size(v) + 1).sum::<usize>(),
329        Value::Object(obj) => {
330            2 + obj
331                .iter()
332                .map(|(k, v)| k.len() + 3 + estimate_json_size(v) + 1)
333                .sum::<usize>()
334        }
335    }
336}
337
338#[derive(Debug, Clone)]
339pub struct CompiledPath {
340    pub segments: std::sync::Arc<[String]>,
341}
342
343impl CompiledPath {
344    pub fn new(path: &str) -> Self {
345        let segments: Vec<String> = path.split('.').map(|s| s.to_string()).collect();
346        CompiledPath {
347            segments: segments.into(),
348        }
349    }
350
351    fn segments(&self) -> &[String] {
352        &self.segments
353    }
354}
355
356/// Represents the type of change made to a field for granular dirty tracking.
357/// This enables emitting only the actual changes rather than entire field values.
358#[derive(Debug, Clone)]
359pub enum FieldChange {
360    /// Field was replaced with a new value (emit the full value from state)
361    Replaced,
362    /// Items were appended to an array field (emit only the new items)
363    Appended(Vec<Value>),
364}
365
366/// Tracks field modifications during handler execution with granular change information.
367/// This replaces the simple HashSet<String> approach to enable delta-only emissions.
368#[derive(Debug, Clone, Default)]
369pub struct DirtyTracker {
370    changes: HashMap<String, FieldChange>,
371}
372
373impl DirtyTracker {
374    /// Create a new empty DirtyTracker
375    pub fn new() -> Self {
376        Self {
377            changes: HashMap::new(),
378        }
379    }
380
381    /// Mark a field as replaced (full value will be emitted)
382    pub fn mark_replaced(&mut self, path: &str) {
383        // If there was an append, it's now superseded by a full replacement
384        self.changes.insert(path.to_string(), FieldChange::Replaced);
385    }
386
387    /// Record an appended value for a field
388    pub fn mark_appended(&mut self, path: &str, value: Value) {
389        match self.changes.get_mut(path) {
390            Some(FieldChange::Appended(values)) => {
391                // Add to existing appended values
392                values.push(value);
393            }
394            Some(FieldChange::Replaced) => {
395                // Field was already replaced, keep it as replaced
396                // (the full value including the append will be emitted)
397            }
398            None => {
399                // First append to this field
400                self.changes
401                    .insert(path.to_string(), FieldChange::Appended(vec![value]));
402            }
403        }
404    }
405
406    /// Check if there are any changes tracked
407    pub fn is_empty(&self) -> bool {
408        self.changes.is_empty()
409    }
410
411    /// Get the number of changed fields
412    pub fn len(&self) -> usize {
413        self.changes.len()
414    }
415
416    /// Iterate over all changes
417    pub fn iter(&self) -> impl Iterator<Item = (&String, &FieldChange)> {
418        self.changes.iter()
419    }
420
421    /// Get a set of all dirty field paths (for backward compatibility)
422    pub fn dirty_paths(&self) -> HashSet<String> {
423        self.changes.keys().cloned().collect()
424    }
425
426    /// Consume the tracker and return the changes map
427    pub fn into_changes(self) -> HashMap<String, FieldChange> {
428        self.changes
429    }
430
431    /// Get a reference to the changes map
432    pub fn changes(&self) -> &HashMap<String, FieldChange> {
433        &self.changes
434    }
435
436    /// Get paths that were appended (not replaced)
437    pub fn appended_paths(&self) -> Vec<String> {
438        self.changes
439            .iter()
440            .filter_map(|(path, change)| match change {
441                FieldChange::Appended(_) => Some(path.clone()),
442                FieldChange::Replaced => None,
443            })
444            .collect()
445    }
446}
447
448pub struct VmContext {
449    registers: Vec<RegisterValue>,
450    states: HashMap<u32, StateTable>,
451    debugger: Option<Arc<dyn VmDebugger>>,
452    pub instructions_executed: u64,
453    pub cache_hits: u64,
454    path_cache: HashMap<String, CompiledPath>,
455    pub pda_cache_hits: u64,
456    pub pda_cache_misses: u64,
457    pub pending_queue_size: u64,
458    resolver_requests: VecDeque<ResolverRequest>,
459    resolver_pending: HashMap<String, PendingResolverEntry>,
460    resolver_cache: LruCache<String, ResolverCacheEntry>,
461    pub resolver_cache_hits: u64,
462    pub resolver_cache_misses: u64,
463    current_context: Option<UpdateContext>,
464    warnings: Vec<String>,
465    last_pda_lookup_miss: Option<String>,
466    last_lookup_index_miss: Option<String>,
467    last_pda_registered: Option<String>,
468    last_lookup_index_keys: Vec<String>,
469    pending_pda_reprocess_updates: Vec<PendingAccountUpdate>,
470    scheduled_callbacks: Vec<(u64, ScheduledCallback)>,
471}
472
473#[derive(Debug)]
474pub struct LookupIndex {
475    index: std::sync::Mutex<LruCache<String, Value>>,
476}
477
478impl LookupIndex {
479    pub fn new() -> Self {
480        Self::with_capacity(DEFAULT_MAX_LOOKUP_INDEX_ENTRIES)
481    }
482
483    pub fn with_capacity(capacity: usize) -> Self {
484        LookupIndex {
485            index: std::sync::Mutex::new(LruCache::new(
486                NonZeroUsize::new(capacity).expect("capacity must be > 0"),
487            )),
488        }
489    }
490
491    pub fn lookup(&self, lookup_value: &Value) -> Option<Value> {
492        let key = value_to_cache_key(lookup_value);
493        self.index.lock().unwrap().get(&key).cloned()
494    }
495
496    pub fn insert(&self, lookup_value: Value, primary_key: Value) {
497        let key = value_to_cache_key(&lookup_value);
498        self.index.lock().unwrap().put(key, primary_key);
499    }
500
501    /// Remove an entry by lookup value.  Used when a PDA mapping changes to
502    /// clear stale entries that would otherwise shadow the updated PDA mapping.
503    pub fn remove(&self, lookup_value: &Value) {
504        let key = value_to_cache_key(lookup_value);
505        self.index.lock().unwrap().pop(&key);
506    }
507
508    pub fn len(&self) -> usize {
509        self.index.lock().unwrap().len()
510    }
511
512    pub fn is_empty(&self) -> bool {
513        self.index.lock().unwrap().is_empty()
514    }
515}
516
517impl Default for LookupIndex {
518    fn default() -> Self {
519        Self::new()
520    }
521}
522
523fn value_to_cache_key(value: &Value) -> String {
524    match value {
525        Value::String(s) => s.clone(),
526        Value::Number(n) => n.to_string(),
527        Value::Bool(b) => b.to_string(),
528        Value::Null => "null".to_string(),
529        _ => serde_json::to_string(value).unwrap_or_else(|_| "unknown".to_string()),
530    }
531}
532
533pub(crate) fn resolver_cache_key(resolver: &ResolverType, input: &Value) -> String {
534    match resolver {
535        ResolverType::Token => format!("token:{}", value_to_cache_key(input)),
536        ResolverType::Url(config) => {
537            let method = match config.method {
538                ast::HttpMethod::Get => "get",
539                ast::HttpMethod::Post => "post",
540            };
541            format!("url:{}:{}", method, value_to_cache_key(input))
542        }
543    }
544}
545
546#[derive(Debug)]
547pub struct TemporalIndex {
548    index: std::sync::Mutex<LruCache<String, Vec<(Value, i64)>>>,
549}
550
551impl Default for TemporalIndex {
552    fn default() -> Self {
553        Self::new()
554    }
555}
556
557impl TemporalIndex {
558    pub fn new() -> Self {
559        Self::with_capacity(DEFAULT_MAX_TEMPORAL_INDEX_KEYS)
560    }
561
562    pub fn with_capacity(capacity: usize) -> Self {
563        TemporalIndex {
564            index: std::sync::Mutex::new(LruCache::new(
565                NonZeroUsize::new(capacity).expect("capacity must be > 0"),
566            )),
567        }
568    }
569
570    pub fn lookup(&self, lookup_value: &Value, timestamp: i64) -> Option<Value> {
571        let key = value_to_cache_key(lookup_value);
572        let mut cache = self.index.lock().unwrap();
573        if let Some(entries) = cache.get(&key) {
574            for i in (0..entries.len()).rev() {
575                if entries[i].1 <= timestamp {
576                    return Some(entries[i].0.clone());
577                }
578            }
579        }
580        None
581    }
582
583    pub fn lookup_latest(&self, lookup_value: &Value) -> Option<Value> {
584        let key = value_to_cache_key(lookup_value);
585        let mut cache = self.index.lock().unwrap();
586        if let Some(entries) = cache.get(&key) {
587            if let Some(last) = entries.last() {
588                return Some(last.0.clone());
589            }
590        }
591        None
592    }
593
594    pub fn insert(&self, lookup_value: Value, primary_key: Value, timestamp: i64) {
595        let key = value_to_cache_key(&lookup_value);
596        let mut cache = self.index.lock().unwrap();
597
598        let entries = cache.get_or_insert_mut(key, Vec::new);
599        entries.push((primary_key, timestamp));
600        entries.sort_by_key(|(_, ts)| *ts);
601
602        let cutoff = timestamp - TEMPORAL_HISTORY_TTL_SECONDS;
603        entries.retain(|(_, ts)| *ts >= cutoff);
604
605        if entries.len() > MAX_TEMPORAL_ENTRIES_PER_KEY {
606            let excess = entries.len() - MAX_TEMPORAL_ENTRIES_PER_KEY;
607            entries.drain(0..excess);
608        }
609    }
610
611    pub fn len(&self) -> usize {
612        self.index.lock().unwrap().len()
613    }
614
615    pub fn is_empty(&self) -> bool {
616        self.index.lock().unwrap().is_empty()
617    }
618
619    pub fn total_entries(&self) -> usize {
620        self.index
621            .lock()
622            .unwrap()
623            .iter()
624            .map(|(_, entries)| entries.len())
625            .sum()
626    }
627
628    pub fn cleanup_expired(&self, cutoff_timestamp: i64) -> usize {
629        let mut cache = self.index.lock().unwrap();
630        let mut total_removed = 0;
631
632        for (_, entries) in cache.iter_mut() {
633            let original_len = entries.len();
634            entries.retain(|(_, ts)| *ts >= cutoff_timestamp);
635            total_removed += original_len - entries.len();
636        }
637
638        total_removed
639    }
640}
641
642#[derive(Debug)]
643pub struct PdaReverseLookup {
644    // Maps: PDA address -> seed value (e.g., bonding_curve_addr -> mint)
645    index: LruCache<String, String>,
646}
647
648impl PdaReverseLookup {
649    pub fn new(capacity: usize) -> Self {
650        PdaReverseLookup {
651            index: LruCache::new(NonZeroUsize::new(capacity).unwrap()),
652        }
653    }
654
655    pub fn lookup(&mut self, pda_address: &str) -> Option<String> {
656        self.index.get(pda_address).cloned()
657    }
658
659    pub fn insert(&mut self, pda_address: String, seed_value: String) -> Option<String> {
660        let evicted = if self.index.len() >= self.index.cap().get() {
661            self.index.peek_lru().map(|(k, _)| k.clone())
662        } else {
663            None
664        };
665
666        self.index.put(pda_address, seed_value);
667        evicted
668    }
669
670    pub fn len(&self) -> usize {
671        self.index.len()
672    }
673
674    pub fn is_empty(&self) -> bool {
675        self.index.is_empty()
676    }
677
678    pub fn contains(&self, pda_address: &str) -> bool {
679        self.index.peek(pda_address).is_some()
680    }
681}
682
683/// Input for queueing an account update.
684#[derive(Debug, Clone)]
685pub struct QueuedAccountUpdate {
686    pub pda_address: String,
687    pub account_type: String,
688    pub account_data: Value,
689    pub slot: u64,
690    pub write_version: u64,
691    pub signature: String,
692}
693
694/// Internal representation of a pending account update with queue metadata.
695#[derive(Debug, Clone)]
696pub struct PendingAccountUpdate {
697    pub account_type: String,
698    pub pda_address: String,
699    pub account_data: Value,
700    pub slot: u64,
701    pub write_version: u64,
702    pub signature: String,
703    pub queued_at: i64,
704    /// When true, this update was pulled from the `last_account_data` cache
705    /// during a PDA mapping change. It carries stale data from a previous
706    /// entity mapping and should use `UpdateContext::new_reprocessed` to
707    /// prevent `when` guards from matching stale instruction signatures
708    /// and to skip resolver scheduling.
709    pub is_stale_reprocess: bool,
710}
711
712/// Input for queueing an instruction event when PDA lookup fails.
713#[derive(Debug, Clone)]
714pub struct QueuedInstructionEvent {
715    pub pda_address: String,
716    pub event_type: String,
717    pub event_data: Value,
718    pub slot: u64,
719    pub signature: String,
720}
721
722/// Internal representation of a pending instruction event with queue metadata.
723#[derive(Debug, Clone)]
724pub struct PendingInstructionEvent {
725    pub event_type: String,
726    pub pda_address: String,
727    pub event_data: Value,
728    pub slot: u64,
729    pub signature: String,
730    pub queued_at: i64,
731}
732
733#[derive(Debug, Clone)]
734pub struct DeferredWhenOperation {
735    pub entity_name: String,
736    pub primary_key: Value,
737    pub field_path: String,
738    pub field_value: Value,
739    pub when_instruction: String,
740    pub signature: String,
741    pub slot: u64,
742    pub deferred_at: i64,
743    pub emit: bool,
744}
745
746#[derive(Debug, Clone)]
747pub struct ResolverRequest {
748    pub cache_key: String,
749    pub resolver: ResolverType,
750    pub input: Value,
751}
752
753#[derive(Debug, Clone)]
754pub struct ResolverTarget {
755    pub state_id: u32,
756    pub entity_name: String,
757    pub primary_key: Value,
758    pub extracts: Vec<ResolverExtractSpec>,
759}
760
761#[derive(Debug, Clone)]
762pub struct PendingResolverEntry {
763    pub resolver: ResolverType,
764    pub input: Value,
765    pub targets: Vec<ResolverTarget>,
766    pub queued_at: i64,
767    pub next_retry_at: Instant,
768    pub retry_count: u32,
769    pub queued: bool,
770    pub in_flight: bool,
771}
772
773#[derive(Debug, Clone, PartialEq)]
774pub struct ScheduledCallback {
775    pub state_id: u32,
776    pub entity_name: String,
777    pub primary_key: Value,
778    pub resolver: ResolverType,
779    pub url_template: Option<Vec<ast::UrlTemplatePart>>,
780    pub input_value: Option<Value>,
781    pub input_path: Option<String>,
782    pub condition: Option<ast::ResolverCondition>,
783    pub strategy: ResolveStrategy,
784    pub extracts: Vec<ResolverExtractSpec>,
785    pub retry_count: u32,
786}
787
788impl PendingResolverEntry {
789    fn add_target(&mut self, target: ResolverTarget) {
790        if let Some(existing) = self.targets.iter_mut().find(|t| {
791            t.state_id == target.state_id
792                && t.entity_name == target.entity_name
793                && t.primary_key == target.primary_key
794        }) {
795            let mut seen = HashSet::new();
796            for extract in &existing.extracts {
797                seen.insert((extract.target_path.clone(), extract.source_path.clone()));
798            }
799
800            for extract in target.extracts {
801                let key = (extract.target_path.clone(), extract.source_path.clone());
802                if seen.insert(key) {
803                    existing.extracts.push(extract);
804                }
805            }
806        } else {
807            self.targets.push(target);
808        }
809    }
810
811    fn ready_to_queue(&self) -> bool {
812        !self.queued && !self.in_flight && Instant::now() >= self.next_retry_at
813    }
814}
815
816#[derive(Debug, Clone)]
817pub struct PendingQueueStats {
818    pub total_updates: usize,
819    pub unique_pdas: usize,
820    pub oldest_age_seconds: i64,
821    pub largest_pda_queue_size: usize,
822    pub estimated_memory_bytes: usize,
823}
824
825#[derive(Debug, Clone, Default)]
826pub struct VmMemoryStats {
827    pub state_table_entity_count: usize,
828    pub state_table_max_entries: usize,
829    pub state_table_at_capacity: bool,
830    pub lookup_index_count: usize,
831    pub lookup_index_total_entries: usize,
832    pub temporal_index_count: usize,
833    pub temporal_index_total_entries: usize,
834    pub pda_reverse_lookup_count: usize,
835    pub pda_reverse_lookup_total_entries: usize,
836    pub version_tracker_entries: usize,
837    pub pending_queue_stats: Option<PendingQueueStats>,
838    pub path_cache_size: usize,
839}
840
841#[derive(Debug, Clone, Default)]
842pub struct CleanupResult {
843    pub pending_updates_removed: usize,
844    pub temporal_entries_removed: usize,
845}
846
847#[derive(Debug, Clone)]
848pub struct CapacityWarning {
849    pub current_entries: usize,
850    pub max_entries: usize,
851    pub entries_over_limit: usize,
852}
853
854#[derive(Debug, Clone)]
855pub struct StateTableConfig {
856    pub max_entries: usize,
857    pub max_array_length: usize,
858}
859
860impl Default for StateTableConfig {
861    fn default() -> Self {
862        Self {
863            max_entries: DEFAULT_MAX_STATE_TABLE_ENTRIES,
864            max_array_length: DEFAULT_MAX_ARRAY_LENGTH,
865        }
866    }
867}
868
869#[derive(Debug)]
870pub struct VersionTracker {
871    cache: std::sync::Mutex<LruCache<String, (u64, u64)>>,
872}
873
874impl VersionTracker {
875    pub fn new() -> Self {
876        Self::with_capacity(DEFAULT_MAX_VERSION_TRACKER_ENTRIES)
877    }
878
879    pub fn with_capacity(capacity: usize) -> Self {
880        VersionTracker {
881            cache: std::sync::Mutex::new(LruCache::new(
882                NonZeroUsize::new(capacity).expect("capacity must be > 0"),
883            )),
884        }
885    }
886
887    fn make_key(primary_key: &Value, event_type: &str) -> String {
888        format!("{}:{}", primary_key, event_type)
889    }
890
891    pub fn get(&self, primary_key: &Value, event_type: &str) -> Option<(u64, u64)> {
892        let key = Self::make_key(primary_key, event_type);
893        self.cache.lock().unwrap().get(&key).copied()
894    }
895
896    pub fn insert(&self, primary_key: &Value, event_type: &str, slot: u64, ordering_value: u64) {
897        let key = Self::make_key(primary_key, event_type);
898        self.cache.lock().unwrap().put(key, (slot, ordering_value));
899    }
900
901    pub fn len(&self) -> usize {
902        self.cache.lock().unwrap().len()
903    }
904
905    pub fn is_empty(&self) -> bool {
906        self.cache.lock().unwrap().is_empty()
907    }
908}
909
910impl Default for VersionTracker {
911    fn default() -> Self {
912        Self::new()
913    }
914}
915
916#[derive(Debug)]
917pub struct StateTable {
918    pub data: DashMap<Value, Value>,
919    access_times: DashMap<Value, i64>,
920    pub lookup_indexes: HashMap<String, LookupIndex>,
921    pub temporal_indexes: HashMap<String, TemporalIndex>,
922    pub pda_reverse_lookups: HashMap<String, PdaReverseLookup>,
923    pub pending_updates: DashMap<String, Vec<PendingAccountUpdate>>,
924    pub pending_instruction_events: DashMap<String, Vec<PendingInstructionEvent>>,
925    /// Cache of the most recent account data per PDA address.  When a PDA
926    /// mapping changes (same PDA, different seed) the cached data is returned
927    /// for reprocessing so cross-account Lookup handlers resolve to the new key.
928    pub last_account_data: DashMap<String, PendingAccountUpdate>,
929    version_tracker: VersionTracker,
930    instruction_dedup_cache: VersionTracker,
931    config: StateTableConfig,
932    #[cfg_attr(not(feature = "otel"), allow(dead_code))]
933    entity_name: String,
934    pub recent_tx_instructions:
935        std::sync::Mutex<lru::LruCache<String, std::collections::HashSet<String>>>,
936    pub deferred_when_ops: DashMap<(String, String), Vec<DeferredWhenOperation>>,
937}
938
939impl StateTable {
940    pub fn is_at_capacity(&self) -> bool {
941        self.data.len() >= self.config.max_entries
942    }
943
944    pub fn entries_over_limit(&self) -> usize {
945        self.data.len().saturating_sub(self.config.max_entries)
946    }
947
948    pub fn max_array_length(&self) -> usize {
949        self.config.max_array_length
950    }
951
952    fn touch(&self, key: &Value) {
953        let now = std::time::SystemTime::now()
954            .duration_since(std::time::UNIX_EPOCH)
955            .unwrap()
956            .as_secs() as i64;
957        self.access_times.insert(key.clone(), now);
958    }
959
960    fn evict_lru(&self, count: usize) -> usize {
961        if count == 0 || self.data.is_empty() {
962            return 0;
963        }
964
965        let mut entries: Vec<(Value, i64)> = self
966            .access_times
967            .iter()
968            .map(|entry| (entry.key().clone(), *entry.value()))
969            .collect();
970
971        entries.sort_by_key(|(_, ts)| *ts);
972
973        let to_evict: Vec<Value> = entries.iter().take(count).map(|(k, _)| k.clone()).collect();
974
975        let mut evicted = 0;
976        for key in to_evict {
977            self.data.remove(&key);
978            self.access_times.remove(&key);
979            evicted += 1;
980        }
981
982        #[cfg(feature = "otel")]
983        if evicted > 0 {
984            crate::vm_metrics::record_state_table_eviction(evicted as u64, &self.entity_name);
985        }
986
987        evicted
988    }
989
990    pub fn insert_with_eviction(&self, key: Value, value: Value) {
991        if self.data.len() >= self.config.max_entries && !self.data.contains_key(&key) {
992            #[cfg(feature = "otel")]
993            crate::vm_metrics::record_state_table_at_capacity(&self.entity_name);
994            let to_evict = (self.data.len() + 1).saturating_sub(self.config.max_entries);
995            self.evict_lru(to_evict.max(1));
996        }
997        self.data.insert(key.clone(), value);
998        self.touch(&key);
999    }
1000
1001    pub fn get_and_touch(&self, key: &Value) -> Option<Value> {
1002        let result = self.data.get(key).map(|v| v.clone());
1003        if result.is_some() {
1004            self.touch(key);
1005        }
1006        result
1007    }
1008
1009    /// Check if an update is fresh and update the version tracker.
1010    /// Returns true if the update should be processed (is fresh).
1011    /// Returns false if the update is stale and should be skipped.
1012    ///
1013    /// Comparison is lexicographic on (slot, ordering_value):
1014    /// (100, 5) > (100, 3) > (99, 999)
1015    pub fn is_fresh_update(
1016        &self,
1017        primary_key: &Value,
1018        event_type: &str,
1019        slot: u64,
1020        ordering_value: u64,
1021    ) -> bool {
1022        let dominated = self
1023            .version_tracker
1024            .get(primary_key, event_type)
1025            .map(|(last_slot, last_version)| (slot, ordering_value) <= (last_slot, last_version))
1026            .unwrap_or(false);
1027
1028        if dominated {
1029            return false;
1030        }
1031
1032        self.version_tracker
1033            .insert(primary_key, event_type, slot, ordering_value);
1034        true
1035    }
1036
1037    /// Check if an instruction is a duplicate of one we've seen recently.
1038    /// Returns true if this exact instruction has been seen before (is a duplicate).
1039    /// Returns false if this is a new instruction that should be processed.
1040    ///
1041    /// Unlike account updates, instructions don't use recency checks - all
1042    /// unique instructions are processed. Only exact duplicates are skipped.
1043    /// Uses a smaller cache capacity for shorter effective TTL.
1044    pub fn is_duplicate_instruction(
1045        &self,
1046        primary_key: &Value,
1047        event_type: &str,
1048        slot: u64,
1049        txn_index: u64,
1050    ) -> bool {
1051        // Check if we've seen this exact instruction before
1052        let is_duplicate = self
1053            .instruction_dedup_cache
1054            .get(primary_key, event_type)
1055            .map(|(last_slot, last_txn_index)| slot == last_slot && txn_index == last_txn_index)
1056            .unwrap_or(false);
1057
1058        if is_duplicate {
1059            return true;
1060        }
1061
1062        // Record this instruction for deduplication
1063        self.instruction_dedup_cache
1064            .insert(primary_key, event_type, slot, txn_index);
1065        false
1066    }
1067}
1068
1069impl VmContext {
1070    pub fn new() -> Self {
1071        let mut vm = VmContext {
1072            registers: vec![Value::Null; 256],
1073            states: HashMap::new(),
1074            debugger: None,
1075            instructions_executed: 0,
1076            cache_hits: 0,
1077            path_cache: HashMap::new(),
1078            pda_cache_hits: 0,
1079            pda_cache_misses: 0,
1080            pending_queue_size: 0,
1081            resolver_requests: VecDeque::new(),
1082            resolver_pending: HashMap::new(),
1083            resolver_cache: LruCache::new(resolver_cache_capacity()),
1084            resolver_cache_hits: 0,
1085            resolver_cache_misses: 0,
1086            current_context: None,
1087            warnings: Vec::new(),
1088            last_pda_lookup_miss: None,
1089            last_lookup_index_miss: None,
1090            last_pda_registered: None,
1091            last_lookup_index_keys: Vec::new(),
1092            pending_pda_reprocess_updates: Vec::new(),
1093            scheduled_callbacks: Vec::new(),
1094        };
1095        vm.states.insert(
1096            0,
1097            StateTable {
1098                data: DashMap::new(),
1099                access_times: DashMap::new(),
1100                lookup_indexes: HashMap::new(),
1101                temporal_indexes: HashMap::new(),
1102                pda_reverse_lookups: HashMap::new(),
1103                pending_updates: DashMap::new(),
1104                pending_instruction_events: DashMap::new(),
1105                last_account_data: DashMap::new(),
1106                version_tracker: VersionTracker::new(),
1107                instruction_dedup_cache: VersionTracker::with_capacity(
1108                    DEFAULT_MAX_INSTRUCTION_DEDUP_ENTRIES,
1109                ),
1110                config: StateTableConfig::default(),
1111                entity_name: String::new(),
1112                recent_tx_instructions: std::sync::Mutex::new(LruCache::new(
1113                    NonZeroUsize::new(1000).unwrap(),
1114                )),
1115                deferred_when_ops: DashMap::new(),
1116            },
1117        );
1118
1119        vm
1120    }
1121
1122    pub fn set_debugger(&mut self, debugger: Arc<dyn VmDebugger>) {
1123        self.debugger = Some(debugger);
1124    }
1125
1126    pub fn clear_debugger(&mut self) {
1127        self.debugger = None;
1128    }
1129
1130    fn emit_debug<F>(&self, builder: F)
1131    where
1132        F: FnOnce() -> VmDebugEvent,
1133    {
1134        if let Some(debugger) = &self.debugger {
1135            debugger.record(builder());
1136        }
1137    }
1138
1139    fn is_debug_enabled(&self) -> bool {
1140        self.debugger.is_some()
1141    }
1142
1143    fn take_pending_pda_reprocess_updates(&mut self) -> Vec<PendingAccountUpdate> {
1144        std::mem::take(&mut self.pending_pda_reprocess_updates)
1145    }
1146
1147    /// Create a new VmContext specifically for multi-entity operation.
1148    pub fn new_multi_entity() -> Self {
1149        VmContext {
1150            registers: vec![Value::Null; 256],
1151            states: HashMap::new(),
1152            debugger: None,
1153            instructions_executed: 0,
1154            cache_hits: 0,
1155            path_cache: HashMap::new(),
1156            pda_cache_hits: 0,
1157            pda_cache_misses: 0,
1158            pending_queue_size: 0,
1159            resolver_requests: VecDeque::new(),
1160            resolver_pending: HashMap::new(),
1161            resolver_cache: LruCache::new(resolver_cache_capacity()),
1162            resolver_cache_hits: 0,
1163            resolver_cache_misses: 0,
1164            current_context: None,
1165            warnings: Vec::new(),
1166            last_pda_lookup_miss: None,
1167            last_lookup_index_miss: None,
1168            last_pda_registered: None,
1169            last_lookup_index_keys: Vec::new(),
1170            pending_pda_reprocess_updates: Vec::new(),
1171            scheduled_callbacks: Vec::new(),
1172        }
1173    }
1174
1175    pub fn new_with_config(state_config: StateTableConfig) -> Self {
1176        let mut vm = VmContext {
1177            registers: vec![Value::Null; 256],
1178            states: HashMap::new(),
1179            debugger: None,
1180            instructions_executed: 0,
1181            cache_hits: 0,
1182            path_cache: HashMap::new(),
1183            pda_cache_hits: 0,
1184            pda_cache_misses: 0,
1185            pending_queue_size: 0,
1186            resolver_requests: VecDeque::new(),
1187            resolver_pending: HashMap::new(),
1188            resolver_cache: LruCache::new(resolver_cache_capacity()),
1189            resolver_cache_hits: 0,
1190            resolver_cache_misses: 0,
1191            current_context: None,
1192            warnings: Vec::new(),
1193            last_pda_lookup_miss: None,
1194            last_lookup_index_miss: None,
1195            last_pda_registered: None,
1196            last_lookup_index_keys: Vec::new(),
1197            pending_pda_reprocess_updates: Vec::new(),
1198            scheduled_callbacks: Vec::new(),
1199        };
1200        vm.states.insert(
1201            0,
1202            StateTable {
1203                data: DashMap::new(),
1204                access_times: DashMap::new(),
1205                lookup_indexes: HashMap::new(),
1206                temporal_indexes: HashMap::new(),
1207                pda_reverse_lookups: HashMap::new(),
1208                pending_updates: DashMap::new(),
1209                pending_instruction_events: DashMap::new(),
1210                last_account_data: DashMap::new(),
1211                version_tracker: VersionTracker::new(),
1212                instruction_dedup_cache: VersionTracker::with_capacity(
1213                    DEFAULT_MAX_INSTRUCTION_DEDUP_ENTRIES,
1214                ),
1215                config: state_config,
1216                entity_name: "default".to_string(),
1217                recent_tx_instructions: std::sync::Mutex::new(LruCache::new(
1218                    NonZeroUsize::new(1000).unwrap(),
1219                )),
1220                deferred_when_ops: DashMap::new(),
1221            },
1222        );
1223        vm
1224    }
1225
1226    pub fn take_resolver_requests(&mut self) -> Vec<ResolverRequest> {
1227        let requests: Vec<ResolverRequest> = self.resolver_requests.drain(..).collect();
1228        for request in &requests {
1229            if let Some(entry) = self.resolver_pending.get_mut(&request.cache_key) {
1230                entry.queued = false;
1231                entry.in_flight = true;
1232            }
1233        }
1234        requests
1235    }
1236
1237    pub fn take_scheduled_callbacks(&mut self) -> Vec<(u64, ScheduledCallback)> {
1238        std::mem::take(&mut self.scheduled_callbacks)
1239    }
1240
1241    pub fn get_entity_state(&self, state_id: u32, key: &Value) -> Option<Value> {
1242        self.states.get(&state_id)?.get_and_touch(key)
1243    }
1244
1245    pub fn snapshot_state_table(&self, state_id: u32) -> Vec<(Value, Value)> {
1246        self.states
1247            .get(&state_id)
1248            .map(|state| {
1249                state
1250                    .data
1251                    .iter()
1252                    .map(|entry| (entry.key().clone(), entry.value().clone()))
1253                    .collect()
1254            })
1255            .unwrap_or_default()
1256    }
1257
1258    pub fn restore_resolver_requests(&mut self, requests: Vec<ResolverRequest>) {
1259        if requests.is_empty() {
1260            return;
1261        }
1262
1263        for request in &requests {
1264            if let Some(entry) = self.resolver_pending.get_mut(&request.cache_key) {
1265                entry.in_flight = false;
1266                entry.queued = true;
1267                entry.next_retry_at = Instant::now();
1268            }
1269        }
1270
1271        self.resolver_requests.extend(requests);
1272    }
1273
1274    pub(crate) fn get_cached_resolver_value(
1275        &mut self,
1276        cache_key: &str,
1277    ) -> Option<CachedResolverValue> {
1278        let cached = self.resolver_cache.get(cache_key).cloned();
1279
1280        match cached {
1281            Some(entry) if entry.cached_at.elapsed() <= entry.ttl => {
1282                self.resolver_cache_hits += 1;
1283                Some(match entry.value {
1284                    ResolverCacheValue::Resolved(value) => CachedResolverValue::Resolved(value),
1285                    ResolverCacheValue::Negative => CachedResolverValue::Negative,
1286                })
1287            }
1288            Some(_) => {
1289                self.resolver_cache.pop(cache_key);
1290                self.resolver_cache_misses += 1;
1291                None
1292            }
1293            None => {
1294                self.resolver_cache_misses += 1;
1295                None
1296            }
1297        }
1298    }
1299
1300    fn cache_resolver_value(
1301        &mut self,
1302        resolver: &ResolverType,
1303        input: &Value,
1304        resolved_value: &Value,
1305    ) {
1306        self.resolver_cache.put(
1307            resolver_cache_key(resolver, input),
1308            ResolverCacheEntry {
1309                value: ResolverCacheValue::Resolved(resolved_value.clone()),
1310                cached_at: Instant::now(),
1311                ttl: resolver_cache_ttl(),
1312            },
1313        );
1314    }
1315
1316    pub(crate) fn cache_negative_resolver_value(&mut self, resolver: &ResolverType, input: &Value) {
1317        self.resolver_cache.put(
1318            resolver_cache_key(resolver, input),
1319            ResolverCacheEntry {
1320                value: ResolverCacheValue::Negative,
1321                cached_at: Instant::now(),
1322                ttl: negative_resolver_cache_ttl(),
1323            },
1324        );
1325    }
1326
1327    pub(crate) fn schedule_resolver_retry(&mut self, cache_key: &str) -> bool {
1328        if let Some(entry) = self.resolver_pending.get_mut(cache_key) {
1329            entry.retry_count = entry.retry_count.saturating_add(1);
1330            entry.next_retry_at = Instant::now() + resolver_retry_backoff();
1331            entry.in_flight = false;
1332            entry.queued = false;
1333            return true;
1334        }
1335        false
1336    }
1337
1338    pub(crate) fn handle_token_resolver_miss(
1339        &mut self,
1340        cache_key: &str,
1341    ) -> TokenResolverMissAction {
1342        let should_retry = self
1343            .resolver_pending
1344            .get(cache_key)
1345            .map(|entry| entry.retry_count < resolver_max_retries())
1346            .unwrap_or(false);
1347
1348        if should_retry && self.schedule_resolver_retry(cache_key) {
1349            return TokenResolverMissAction::RetryScheduled;
1350        }
1351
1352        if let Some(entry) = self.drop_pending_resolver_entry(cache_key) {
1353            self.cache_negative_resolver_value(&entry.resolver, &entry.input);
1354            return TokenResolverMissAction::NegativeCached;
1355        }
1356
1357        TokenResolverMissAction::Dropped
1358    }
1359
1360    pub(crate) fn drop_pending_resolver_entry(
1361        &mut self,
1362        cache_key: &str,
1363    ) -> Option<PendingResolverEntry> {
1364        self.resolver_pending.remove(cache_key)
1365    }
1366
1367    pub fn apply_resolver_result(
1368        &mut self,
1369        bytecode: &MultiEntityBytecode,
1370        cache_key: &str,
1371        resolved_value: Value,
1372    ) -> Result<Vec<Mutation>> {
1373        let entry = match self.resolver_pending.remove(cache_key) {
1374            Some(entry) => entry,
1375            None => return Ok(Vec::new()),
1376        };
1377
1378        self.cache_resolver_value(&entry.resolver, &entry.input, &resolved_value);
1379
1380        let mut mutations = Vec::new();
1381
1382        for target in entry.targets {
1383            if target.primary_key.is_null() {
1384                continue;
1385            }
1386
1387            let entity_bytecode = match bytecode.entities.get(&target.entity_name) {
1388                Some(bc) => bc,
1389                None => continue,
1390            };
1391
1392            let state = match self.states.get(&target.state_id) {
1393                Some(s) => s,
1394                None => continue,
1395            };
1396
1397            let mut entity_state = state
1398                .get_and_touch(&target.primary_key)
1399                .unwrap_or_else(|| json!({}));
1400
1401            let mut dirty_tracker = DirtyTracker::new();
1402            let should_emit = |path: &str| !entity_bytecode.non_emitted_fields.contains(path);
1403
1404            Self::apply_resolver_extractions_to_value(
1405                &mut entity_state,
1406                &resolved_value,
1407                &target.extracts,
1408                &mut dirty_tracker,
1409                &should_emit,
1410            )?;
1411
1412            if let Some(evaluator) = entity_bytecode.computed_fields_evaluator.as_ref() {
1413                let old_values: Vec<_> = entity_bytecode
1414                    .computed_paths
1415                    .iter()
1416                    .map(|path| Self::get_value_at_path(&entity_state, path))
1417                    .collect();
1418
1419                let context_slot = self.current_context.as_ref().and_then(|c| c.slot);
1420                let context_timestamp = self
1421                    .current_context
1422                    .as_ref()
1423                    .map(|c| c.timestamp())
1424                    .unwrap_or_else(|| {
1425                        std::time::SystemTime::now()
1426                            .duration_since(std::time::UNIX_EPOCH)
1427                            .unwrap()
1428                            .as_secs() as i64
1429                    });
1430                let eval_result = evaluator(&mut entity_state, context_slot, context_timestamp);
1431
1432                if eval_result.is_ok() {
1433                    let mut changed_fields = Vec::new();
1434                    for (path, old_value) in
1435                        entity_bytecode.computed_paths.iter().zip(old_values.iter())
1436                    {
1437                        let new_value = Self::get_value_at_path(&entity_state, path);
1438                        let changed = new_value != *old_value;
1439                        let will_emit = should_emit(path);
1440
1441                        if changed && will_emit {
1442                            dirty_tracker.mark_replaced(path);
1443                            changed_fields.push(path.clone());
1444                        }
1445                    }
1446                }
1447            }
1448
1449            state.insert_with_eviction(target.primary_key.clone(), entity_state.clone());
1450
1451            if dirty_tracker.is_empty() {
1452                continue;
1453            }
1454
1455            let patch = Self::build_partial_state_from_value(&entity_state, &dirty_tracker)?;
1456
1457            mutations.push(Mutation {
1458                export: target.entity_name.clone(),
1459                key: target.primary_key.clone(),
1460                patch,
1461                append: vec![],
1462            });
1463        }
1464
1465        Ok(mutations)
1466    }
1467
1468    pub fn enqueue_resolver_request(
1469        &mut self,
1470        _cache_key: String,
1471        resolver: ResolverType,
1472        input: Value,
1473        target: ResolverTarget,
1474    ) {
1475        let cache_key = resolver_cache_key(&resolver, &input);
1476
1477        if let Some(entry) = self.resolver_pending.get_mut(&cache_key) {
1478            entry.add_target(target);
1479            if entry.ready_to_queue() {
1480                entry.queued = true;
1481                self.resolver_requests.push_back(ResolverRequest {
1482                    cache_key,
1483                    resolver,
1484                    input,
1485                });
1486            }
1487            return;
1488        }
1489
1490        let queued_at = std::time::SystemTime::now()
1491            .duration_since(std::time::UNIX_EPOCH)
1492            .unwrap()
1493            .as_secs() as i64;
1494
1495        self.resolver_pending.insert(
1496            cache_key.clone(),
1497            PendingResolverEntry {
1498                resolver: resolver.clone(),
1499                input: input.clone(),
1500                targets: vec![target],
1501                queued_at,
1502                next_retry_at: Instant::now(),
1503                retry_count: 0,
1504                queued: true,
1505                in_flight: false,
1506            },
1507        );
1508
1509        self.resolver_requests.push_back(ResolverRequest {
1510            cache_key,
1511            resolver,
1512            input,
1513        });
1514    }
1515
1516    fn apply_resolver_extractions_to_value<F>(
1517        state: &mut Value,
1518        resolved_value: &Value,
1519        extracts: &[ResolverExtractSpec],
1520        dirty_tracker: &mut DirtyTracker,
1521        should_emit: &F,
1522    ) -> Result<()>
1523    where
1524        F: Fn(&str) -> bool,
1525    {
1526        for extract in extracts {
1527            let resolved = match extract.source_path.as_deref() {
1528                Some(path) => Self::get_value_at_path(resolved_value, path),
1529                None => Some(resolved_value.clone()),
1530            };
1531
1532            let Some(value) = resolved else {
1533                continue;
1534            };
1535
1536            let value = if let Some(transform) = &extract.transform {
1537                Self::apply_transformation(&value, transform)?
1538            } else {
1539                value
1540            };
1541
1542            Self::set_nested_field_value(state, &extract.target_path, value)?;
1543            if should_emit(&extract.target_path) {
1544                dirty_tracker.mark_replaced(&extract.target_path);
1545            }
1546        }
1547
1548        Ok(())
1549    }
1550
1551    fn build_partial_state_from_value(state: &Value, tracker: &DirtyTracker) -> Result<Value> {
1552        if tracker.is_empty() {
1553            return Ok(json!({}));
1554        }
1555
1556        let mut partial = serde_json::Map::new();
1557
1558        for (path, change) in tracker.iter() {
1559            let segments: Vec<&str> = path.split('.').collect();
1560
1561            let value_to_insert = match change {
1562                FieldChange::Replaced => {
1563                    let mut current = state;
1564                    let mut found = true;
1565
1566                    for segment in &segments {
1567                        match current.get(*segment) {
1568                            Some(v) => current = v,
1569                            None => {
1570                                found = false;
1571                                break;
1572                            }
1573                        }
1574                    }
1575
1576                    if !found {
1577                        continue;
1578                    }
1579                    current.clone()
1580                }
1581                FieldChange::Appended(values) => Value::Array(values.clone()),
1582            };
1583
1584            let mut target = &mut partial;
1585            for (i, segment) in segments.iter().enumerate() {
1586                if i == segments.len() - 1 {
1587                    target.insert(segment.to_string(), value_to_insert.clone());
1588                } else {
1589                    target
1590                        .entry(segment.to_string())
1591                        .or_insert_with(|| json!({}));
1592                    target = target
1593                        .get_mut(*segment)
1594                        .and_then(|v| v.as_object_mut())
1595                        .ok_or("Failed to build nested structure")?;
1596                }
1597            }
1598        }
1599
1600        Ok(Value::Object(partial))
1601    }
1602
1603    /// Get a mutable reference to a state table by ID
1604    /// Returns None if the state ID doesn't exist
1605    pub fn get_state_table_mut(&mut self, state_id: u32) -> Option<&mut StateTable> {
1606        self.states.get_mut(&state_id)
1607    }
1608
1609    /// Get public access to registers (for metrics context)
1610    pub fn registers_mut(&mut self) -> &mut Vec<RegisterValue> {
1611        &mut self.registers
1612    }
1613
1614    /// Get public access to path cache (for metrics context)
1615    pub fn path_cache(&self) -> &HashMap<String, CompiledPath> {
1616        &self.path_cache
1617    }
1618
1619    /// Get the current update context
1620    pub fn current_context(&self) -> Option<&UpdateContext> {
1621        self.current_context.as_ref()
1622    }
1623
1624    /// Set the current update context for computed field evaluation
1625    pub fn set_current_context(&mut self, context: Option<UpdateContext>) {
1626        self.current_context = context;
1627    }
1628
1629    fn add_warning(&mut self, msg: String) {
1630        self.warnings.push(msg);
1631    }
1632
1633    pub fn take_warnings(&mut self) -> Vec<String> {
1634        std::mem::take(&mut self.warnings)
1635    }
1636
1637    pub fn has_warnings(&self) -> bool {
1638        !self.warnings.is_empty()
1639    }
1640
1641    pub fn update_state_from_register(
1642        &mut self,
1643        state_id: u32,
1644        key: Value,
1645        register: Register,
1646    ) -> Result<()> {
1647        let state = self.states.get(&state_id).ok_or("State table not found")?;
1648        let value = self.registers[register].clone();
1649        state.insert_with_eviction(key, value);
1650        Ok(())
1651    }
1652
1653    fn reset_registers(&mut self) {
1654        for reg in &mut self.registers {
1655            *reg = Value::Null;
1656        }
1657    }
1658
1659    /// Extract only the dirty fields from state (public for use by instruction hooks)
1660    pub fn extract_partial_state(
1661        &self,
1662        state_reg: Register,
1663        dirty_fields: &HashSet<String>,
1664    ) -> Result<Value> {
1665        let full_state = &self.registers[state_reg];
1666
1667        if dirty_fields.is_empty() {
1668            return Ok(json!({}));
1669        }
1670
1671        let mut partial = serde_json::Map::new();
1672
1673        for path in dirty_fields {
1674            let segments: Vec<&str> = path.split('.').collect();
1675
1676            let mut current = full_state;
1677            let mut found = true;
1678
1679            for segment in &segments {
1680                match current.get(segment) {
1681                    Some(v) => current = v,
1682                    None => {
1683                        found = false;
1684                        break;
1685                    }
1686                }
1687            }
1688
1689            if !found {
1690                continue;
1691            }
1692
1693            let mut target = &mut partial;
1694            for (i, segment) in segments.iter().enumerate() {
1695                if i == segments.len() - 1 {
1696                    target.insert(segment.to_string(), current.clone());
1697                } else {
1698                    target
1699                        .entry(segment.to_string())
1700                        .or_insert_with(|| json!({}));
1701                    target = target
1702                        .get_mut(*segment)
1703                        .and_then(|v| v.as_object_mut())
1704                        .ok_or("Failed to build nested structure")?;
1705                }
1706            }
1707        }
1708
1709        Ok(Value::Object(partial))
1710    }
1711
1712    /// Extract a patch from state based on the DirtyTracker.
1713    /// For Replaced fields: extracts the full value from state.
1714    /// For Appended fields: emits only the appended values as an array.
1715    pub fn extract_partial_state_with_tracker(
1716        &self,
1717        state_reg: Register,
1718        tracker: &DirtyTracker,
1719    ) -> Result<Value> {
1720        let full_state = &self.registers[state_reg];
1721
1722        if tracker.is_empty() {
1723            return Ok(json!({}));
1724        }
1725
1726        let mut partial = serde_json::Map::new();
1727
1728        for (path, change) in tracker.iter() {
1729            let segments: Vec<&str> = path.split('.').collect();
1730
1731            let value_to_insert = match change {
1732                FieldChange::Replaced => {
1733                    let mut current = full_state;
1734                    let mut found = true;
1735
1736                    for segment in &segments {
1737                        match current.get(*segment) {
1738                            Some(v) => current = v,
1739                            None => {
1740                                found = false;
1741                                break;
1742                            }
1743                        }
1744                    }
1745
1746                    if !found {
1747                        continue;
1748                    }
1749                    current.clone()
1750                }
1751                FieldChange::Appended(values) => Value::Array(values.clone()),
1752            };
1753
1754            let mut target = &mut partial;
1755            for (i, segment) in segments.iter().enumerate() {
1756                if i == segments.len() - 1 {
1757                    target.insert(segment.to_string(), value_to_insert.clone());
1758                } else {
1759                    target
1760                        .entry(segment.to_string())
1761                        .or_insert_with(|| json!({}));
1762                    target = target
1763                        .get_mut(*segment)
1764                        .and_then(|v| v.as_object_mut())
1765                        .ok_or("Failed to build nested structure")?;
1766                }
1767            }
1768        }
1769
1770        Ok(Value::Object(partial))
1771    }
1772
1773    fn get_compiled_path(&mut self, path: &str) -> CompiledPath {
1774        if let Some(compiled) = self.path_cache.get(path) {
1775            self.cache_hits += 1;
1776            #[cfg(feature = "otel")]
1777            crate::vm_metrics::record_path_cache_hit();
1778            return compiled.clone();
1779        }
1780        #[cfg(feature = "otel")]
1781        crate::vm_metrics::record_path_cache_miss();
1782        let compiled = CompiledPath::new(path);
1783        self.path_cache.insert(path.to_string(), compiled.clone());
1784        compiled
1785    }
1786
1787    /// Process an event with optional context metadata
1788    #[cfg_attr(feature = "otel", instrument(
1789        name = "vm.process_event",
1790        skip(self, bytecode, event_value, log),
1791        level = "info",
1792        fields(
1793            event_type = %event_type,
1794            slot = context.as_ref().and_then(|c| c.slot),
1795        )
1796    ))]
1797    pub fn process_event(
1798        &mut self,
1799        bytecode: &MultiEntityBytecode,
1800        event_value: Value,
1801        event_type: &str,
1802        context: Option<&UpdateContext>,
1803        mut log: Option<&mut crate::canonical_log::CanonicalLog>,
1804    ) -> Result<Vec<Mutation>> {
1805        self.current_context = context.cloned();
1806        self.emit_debug(|| VmDebugEvent::ProcessEventStart {
1807            event_type: event_type.to_string(),
1808            context: context.map(|ctx| ctx.to_value()),
1809        });
1810
1811        let mut event_value = event_value;
1812        if let Some(ctx) = context {
1813            if let Some(obj) = event_value.as_object_mut() {
1814                obj.insert("__update_context".to_string(), ctx.to_value());
1815            }
1816        }
1817
1818        let mut all_mutations = Vec::new();
1819
1820        if event_type.ends_with("IxState") && bytecode.when_events.contains(event_type) {
1821            if let Some(ctx) = context {
1822                if let Some(signature) = ctx.signature.clone() {
1823                    let state_ids: Vec<u32> = self.states.keys().cloned().collect();
1824                    for state_id in state_ids {
1825                        if let Some(state) = self.states.get(&state_id) {
1826                            {
1827                                let mut cache = state.recent_tx_instructions.lock().unwrap();
1828                                let entry =
1829                                    cache.get_or_insert_mut(signature.clone(), HashSet::new);
1830                                entry.insert(event_type.to_string());
1831                            }
1832
1833                            let key = (signature.clone(), event_type.to_string());
1834                            if let Some((_, deferred_ops)) = state.deferred_when_ops.remove(&key) {
1835                                tracing::debug!(
1836                                    event_type = %event_type,
1837                                    signature = %signature,
1838                                    deferred_count = deferred_ops.len(),
1839                                    "flushing deferred when-ops"
1840                                );
1841                                for op in deferred_ops {
1842                                    // Look up the entity bytecode to get the computed fields evaluator
1843                                    let (evaluator, computed_paths) = bytecode
1844                                        .entities
1845                                        .get(&op.entity_name)
1846                                        .map(|eb| {
1847                                            (
1848                                                eb.computed_fields_evaluator.as_ref(),
1849                                                eb.computed_paths.as_slice(),
1850                                            )
1851                                        })
1852                                        .unwrap_or((None, &[]));
1853
1854                                    match self.apply_deferred_when_op(
1855                                        state_id,
1856                                        &op,
1857                                        evaluator,
1858                                        Some(computed_paths),
1859                                    ) {
1860                                        Ok(mutations) => all_mutations.extend(mutations),
1861                                        Err(e) => tracing::warn!(
1862                                            "Failed to apply deferred when-op: {}",
1863                                            e
1864                                        ),
1865                                    }
1866                                }
1867                            }
1868                        }
1869                    }
1870                }
1871            }
1872        }
1873
1874        if let Some(entity_names) = bytecode.event_routing.get(event_type) {
1875            for entity_name in entity_names {
1876                if let Some(entity_bytecode) = bytecode.entities.get(entity_name) {
1877                    if let Some(handler) = entity_bytecode.handlers.get(event_type) {
1878                        self.emit_debug(|| VmDebugEvent::HandlerStart {
1879                            entity_name: entity_name.clone(),
1880                            event_type: event_type.to_string(),
1881                            state_id: entity_bytecode.state_id,
1882                        });
1883
1884                        if let Some(ref mut log) = log {
1885                            log.set("entity", entity_name.clone());
1886                            log.inc("handlers", 1);
1887                        }
1888
1889                        let opcodes_before = self.instructions_executed;
1890                        let cache_before = self.cache_hits;
1891                        let pda_hits_before = self.pda_cache_hits;
1892                        let pda_misses_before = self.pda_cache_misses;
1893
1894                        let mutations = self.execute_handler(
1895                            handler,
1896                            &event_value,
1897                            event_type,
1898                            entity_bytecode.state_id,
1899                            entity_name,
1900                            entity_bytecode.computed_fields_evaluator.as_ref(),
1901                            Some(&entity_bytecode.non_emitted_fields),
1902                        )?;
1903                        let direct_mutation_count = mutations.len();
1904
1905                        if let Some(ref mut log) = log {
1906                            log.inc(
1907                                "opcodes",
1908                                (self.instructions_executed - opcodes_before) as i64,
1909                            );
1910                            log.inc("cache_hits", (self.cache_hits - cache_before) as i64);
1911                            log.inc("pda_hits", (self.pda_cache_hits - pda_hits_before) as i64);
1912                            log.inc(
1913                                "pda_misses",
1914                                (self.pda_cache_misses - pda_misses_before) as i64,
1915                            );
1916                        }
1917
1918                        if mutations.is_empty() {
1919                            // CPI events (suffix "CpiEvent") are transaction-scoped like instructions
1920                            // (suffix "IxState") and should be queued the same way when PDA lookup fails.
1921                            let is_tx_event =
1922                                event_type.ends_with("IxState") || event_type.ends_with("CpiEvent");
1923                            if let Some(missed_pda) = self.take_last_pda_lookup_miss() {
1924                                if is_tx_event {
1925                                    let slot = context.and_then(|c| c.slot).unwrap_or(0);
1926                                    let signature = context
1927                                        .and_then(|c| c.signature.clone())
1928                                        .unwrap_or_default();
1929                                    let _ = self.queue_instruction_event(
1930                                        entity_bytecode.state_id,
1931                                        QueuedInstructionEvent {
1932                                            pda_address: missed_pda.clone(),
1933                                            event_type: event_type.to_string(),
1934                                            event_data: event_value.clone(),
1935                                            slot,
1936                                            signature,
1937                                        },
1938                                    );
1939                                    self.emit_debug(|| VmDebugEvent::QueueAction {
1940                                        entity_name: entity_name.clone(),
1941                                        event_type: event_type.to_string(),
1942                                        queue_kind: "instruction_pda_lookup_miss".to_string(),
1943                                        lookup_value: missed_pda,
1944                                    });
1945                                } else {
1946                                    // Queue account updates (e.g. BondingCurve) when PDA
1947                                    // reverse lookup fails. These will be flushed when an
1948                                    // instruction registers the PDA mapping via
1949                                    // UpdateLookupIndex.
1950                                    let slot = context.and_then(|c| c.slot).unwrap_or(0);
1951                                    let signature = context
1952                                        .and_then(|c| c.signature.clone())
1953                                        .unwrap_or_default();
1954                                    if let Some(write_version) =
1955                                        context.and_then(|c| c.write_version)
1956                                    {
1957                                        let _ = self.queue_account_update(
1958                                            entity_bytecode.state_id,
1959                                            QueuedAccountUpdate {
1960                                                pda_address: missed_pda.clone(),
1961                                                account_type: event_type.to_string(),
1962                                                account_data: event_value.clone(),
1963                                                slot,
1964                                                write_version,
1965                                                signature,
1966                                            },
1967                                        );
1968                                        self.emit_debug(|| VmDebugEvent::QueueAction {
1969                                            entity_name: entity_name.clone(),
1970                                            event_type: event_type.to_string(),
1971                                            queue_kind: "account_pda_lookup_miss".to_string(),
1972                                            lookup_value: missed_pda,
1973                                        });
1974                                    } else {
1975                                        tracing::warn!(
1976                                            event_type = %event_type,
1977                                            "Dropping queued account update: write_version missing from context"
1978                                        );
1979                                    }
1980                                }
1981                            }
1982                            if let Some(missed_lookup) = self.take_last_lookup_index_miss() {
1983                                if !is_tx_event {
1984                                    let slot = context.and_then(|c| c.slot).unwrap_or(0);
1985                                    let signature = context
1986                                        .and_then(|c| c.signature.clone())
1987                                        .unwrap_or_default();
1988                                    if let Some(write_version) =
1989                                        context.and_then(|c| c.write_version)
1990                                    {
1991                                        let _ = self.queue_account_update(
1992                                            entity_bytecode.state_id,
1993                                            QueuedAccountUpdate {
1994                                                pda_address: missed_lookup.clone(),
1995                                                account_type: event_type.to_string(),
1996                                                account_data: event_value.clone(),
1997                                                slot,
1998                                                write_version,
1999                                                signature,
2000                                            },
2001                                        );
2002                                        self.emit_debug(|| VmDebugEvent::QueueAction {
2003                                            entity_name: entity_name.clone(),
2004                                            event_type: event_type.to_string(),
2005                                            queue_kind: "account_lookup_index_miss".to_string(),
2006                                            lookup_value: missed_lookup,
2007                                        });
2008                                    } else {
2009                                        tracing::trace!(
2010                                            event_type = %event_type,
2011                                            "Discarding lookup_index_miss for tx-scoped event (IxState/CpiEvent do not use lookup-index queuing)"
2012                                        );
2013                                    }
2014                                }
2015                            }
2016                        }
2017
2018                        all_mutations.extend(mutations);
2019
2020                        if event_type.ends_with("IxState") || event_type.ends_with("CpiEvent") {
2021                            if let Some(ctx) = context {
2022                                if let Some(ref signature) = ctx.signature {
2023                                    if let Some(state) = self.states.get(&entity_bytecode.state_id)
2024                                    {
2025                                        {
2026                                            let mut cache =
2027                                                state.recent_tx_instructions.lock().unwrap();
2028                                            let entry = cache
2029                                                .get_or_insert_mut(signature.clone(), HashSet::new);
2030                                            entry.insert(event_type.to_string());
2031                                        }
2032
2033                                        let key = (signature.clone(), event_type.to_string());
2034                                        if let Some((_, deferred_ops)) =
2035                                            state.deferred_when_ops.remove(&key)
2036                                        {
2037                                            tracing::debug!(
2038                                                event_type = %event_type,
2039                                                signature = %signature,
2040                                                deferred_count = deferred_ops.len(),
2041                                                "flushing deferred when-ops"
2042                                            );
2043                                            for op in deferred_ops {
2044                                                match self.apply_deferred_when_op(
2045                                                    entity_bytecode.state_id,
2046                                                    &op,
2047                                                    entity_bytecode
2048                                                        .computed_fields_evaluator
2049                                                        .as_ref(),
2050                                                    Some(&entity_bytecode.computed_paths),
2051                                                ) {
2052                                                    Ok(mutations) => {
2053                                                        all_mutations.extend(mutations)
2054                                                    }
2055                                                    Err(e) => {
2056                                                        tracing::warn!(
2057                                                            "Failed to apply deferred when-op: {}",
2058                                                            e
2059                                                        );
2060                                                    }
2061                                                }
2062                                            }
2063                                        }
2064                                    }
2065                                }
2066                            }
2067                        }
2068
2069                        if let Some(registered_pda) = self.take_last_pda_registered() {
2070                            let pending_events = self.flush_pending_instruction_events(
2071                                entity_bytecode.state_id,
2072                                &registered_pda,
2073                            );
2074                            let pending_count = pending_events.len();
2075                            if pending_count > 0 {
2076                                self.emit_debug(|| VmDebugEvent::FlushAction {
2077                                    entity_name: entity_name.clone(),
2078                                    event_type: event_type.to_string(),
2079                                    flush_kind: "pending_instruction_events".to_string(),
2080                                    trigger: registered_pda.clone(),
2081                                    count: pending_count,
2082                                });
2083                            }
2084                            for pending in pending_events {
2085                                if let Some(pending_handler) =
2086                                    entity_bytecode.handlers.get(&pending.event_type)
2087                                {
2088                                    if let Ok(reprocessed_mutations) = self.execute_handler(
2089                                        pending_handler,
2090                                        &pending.event_data,
2091                                        &pending.event_type,
2092                                        entity_bytecode.state_id,
2093                                        entity_name,
2094                                        entity_bytecode.computed_fields_evaluator.as_ref(),
2095                                        Some(&entity_bytecode.non_emitted_fields),
2096                                    ) {
2097                                        all_mutations.extend(reprocessed_mutations);
2098                                    }
2099                                }
2100                            }
2101                        }
2102
2103                        for pending in self.take_pending_pda_reprocess_updates() {
2104                            if let Some(pending_handler) =
2105                                entity_bytecode.handlers.get(&pending.account_type)
2106                            {
2107                                self.current_context = Some(if pending.is_stale_reprocess {
2108                                    UpdateContext::new_reprocessed(
2109                                        pending.slot,
2110                                        pending.write_version,
2111                                    )
2112                                } else {
2113                                    UpdateContext::new_account(
2114                                        pending.slot,
2115                                        pending.signature.clone(),
2116                                        pending.write_version,
2117                                    )
2118                                });
2119                                match self.execute_handler(
2120                                    pending_handler,
2121                                    &pending.account_data,
2122                                    &pending.account_type,
2123                                    entity_bytecode.state_id,
2124                                    entity_name,
2125                                    entity_bytecode.computed_fields_evaluator.as_ref(),
2126                                    Some(&entity_bytecode.non_emitted_fields),
2127                                ) {
2128                                    Ok(reprocessed) => {
2129                                        all_mutations.extend(reprocessed);
2130                                    }
2131                                    Err(e) => {
2132                                        tracing::warn!(
2133                                            error = %e,
2134                                            account_type = %pending.account_type,
2135                                            "PDA remap reprocessing failed"
2136                                        );
2137                                    }
2138                                }
2139                            }
2140                        }
2141
2142                        let lookup_keys = self.take_last_lookup_index_keys();
2143                        if !lookup_keys.is_empty() {
2144                            tracing::info!(
2145                                keys = ?lookup_keys,
2146                                entity = %entity_name,
2147                                "vm.process_event: flushing pending updates for lookup_keys"
2148                            );
2149                        }
2150                        for lookup_key in lookup_keys {
2151                            if let Ok(pending_updates) =
2152                                self.flush_pending_updates(entity_bytecode.state_id, &lookup_key)
2153                            {
2154                                let pending_count = pending_updates.len();
2155                                if pending_count > 0 {
2156                                    self.emit_debug(|| VmDebugEvent::FlushAction {
2157                                        entity_name: entity_name.clone(),
2158                                        event_type: event_type.to_string(),
2159                                        flush_kind: "pending_account_updates".to_string(),
2160                                        trigger: lookup_key.clone(),
2161                                        count: pending_count,
2162                                    });
2163                                }
2164                                for pending in pending_updates {
2165                                    if let Some(pending_handler) =
2166                                        entity_bytecode.handlers.get(&pending.account_type)
2167                                    {
2168                                        self.current_context = Some(UpdateContext::new_account(
2169                                            pending.slot,
2170                                            pending.signature.clone(),
2171                                            pending.write_version,
2172                                        ));
2173                                        match self.execute_handler(
2174                                            pending_handler,
2175                                            &pending.account_data,
2176                                            &pending.account_type,
2177                                            entity_bytecode.state_id,
2178                                            entity_name,
2179                                            entity_bytecode.computed_fields_evaluator.as_ref(),
2180                                            Some(&entity_bytecode.non_emitted_fields),
2181                                        ) {
2182                                            Ok(reprocessed) => {
2183                                                all_mutations.extend(reprocessed);
2184                                            }
2185                                            Err(e) => {
2186                                                tracing::warn!(
2187                                                    error = %e,
2188                                                    account_type = %pending.account_type,
2189                                                    "Flushed event reprocessing failed"
2190                                                );
2191                                            }
2192                                        }
2193                                    }
2194                                }
2195                            }
2196                        }
2197
2198                        self.emit_debug(|| VmDebugEvent::HandlerEnd {
2199                            entity_name: entity_name.clone(),
2200                            event_type: event_type.to_string(),
2201                            mutations: direct_mutation_count,
2202                        });
2203                    } else if let Some(ref mut log) = log {
2204                        log.set("skip_reason", "no_handler");
2205                    }
2206                } else if let Some(ref mut log) = log {
2207                    log.set("skip_reason", "entity_not_found");
2208                }
2209            }
2210        } else if let Some(ref mut log) = log {
2211            log.set("skip_reason", "no_event_routing");
2212        }
2213
2214        let debug_warnings = self.warnings.clone();
2215
2216        if let Some(log) = log {
2217            log.set("mutations", all_mutations.len() as i64);
2218            if let Some(first) = all_mutations.first() {
2219                if let Some(key_str) = first.key.as_str() {
2220                    log.set("primary_key", key_str);
2221                } else if let Some(key_num) = first.key.as_u64() {
2222                    log.set("primary_key", key_num as i64);
2223                }
2224            }
2225            if let Some(state) = self.states.get(&0) {
2226                log.set("state_table_size", state.data.len() as i64);
2227            }
2228
2229            let warnings = self.take_warnings();
2230            if !warnings.is_empty() {
2231                log.set("warnings", warnings.len() as i64);
2232                log.set(
2233                    "warning_messages",
2234                    Value::Array(warnings.into_iter().map(Value::String).collect()),
2235                );
2236                log.set_level(crate::canonical_log::LogLevel::Warn);
2237            }
2238        } else {
2239            self.warnings.clear();
2240        }
2241
2242        self.emit_debug(|| VmDebugEvent::ProcessEventEnd {
2243            event_type: event_type.to_string(),
2244            mutations: all_mutations.len(),
2245            warnings: debug_warnings,
2246        });
2247
2248        if self.instructions_executed.is_multiple_of(1000) {
2249            let state_ids: Vec<u32> = self.states.keys().cloned().collect();
2250            for state_id in state_ids {
2251                let expired = self.cleanup_expired_when_ops(state_id, 60);
2252                if expired > 0 {
2253                    tracing::debug!(
2254                        "Cleaned up {} expired deferred when-ops for state {}",
2255                        expired,
2256                        state_id
2257                    );
2258                }
2259            }
2260        }
2261
2262        Ok(all_mutations)
2263    }
2264
2265    pub fn process_any(
2266        &mut self,
2267        bytecode: &MultiEntityBytecode,
2268        any: prost_types::Any,
2269    ) -> Result<Vec<Mutation>> {
2270        let (event_value, event_type) = bytecode.proto_router.decode(any)?;
2271        self.process_event(bytecode, event_value, &event_type, None, None)
2272    }
2273
2274    #[cfg_attr(feature = "otel", instrument(
2275        name = "vm.execute_handler",
2276        skip(self, handler, event_value, entity_evaluator),
2277        level = "debug",
2278        fields(
2279            event_type = %event_type,
2280            handler_opcodes = handler.len(),
2281        )
2282    ))]
2283    #[allow(clippy::type_complexity, clippy::too_many_arguments)]
2284    fn execute_handler(
2285        &mut self,
2286        handler: &[OpCode],
2287        event_value: &Value,
2288        event_type: &str,
2289        override_state_id: u32,
2290        entity_name: &str,
2291        entity_evaluator: Option<
2292            &Box<dyn Fn(&mut Value, Option<u64>, i64) -> ComputedEvaluatorResult + Send + Sync>,
2293        >,
2294        non_emitted_fields: Option<&HashSet<String>>,
2295    ) -> Result<Vec<Mutation>> {
2296        self.reset_registers();
2297        self.last_pda_lookup_miss = None;
2298
2299        let mut pc: usize = 0;
2300        let mut output = Vec::new();
2301        let mut dirty_tracker = DirtyTracker::new();
2302        let should_emit = |path: &str| {
2303            non_emitted_fields
2304                .map(|fields| !fields.contains(path))
2305                .unwrap_or(true)
2306        };
2307
2308        while pc < handler.len() {
2309            match &handler[pc] {
2310                OpCode::LoadEventField {
2311                    path,
2312                    dest,
2313                    default,
2314                } => {
2315                    let value = self.load_field(event_value, path, default.as_ref())?;
2316                    if self.is_debug_enabled() {
2317                        self.emit_debug(|| VmDebugEvent::LoadEventField {
2318                            entity_name: entity_name.to_string(),
2319                            event_type: event_type.to_string(),
2320                            path: path.segments.clone(),
2321                            value: value.clone(),
2322                        });
2323                    }
2324                    self.registers[*dest] = value;
2325                    pc += 1;
2326                }
2327                OpCode::LoadConstant { value, dest } => {
2328                    self.registers[*dest] = value.clone();
2329                    pc += 1;
2330                }
2331                OpCode::CopyRegister { source, dest } => {
2332                    self.registers[*dest] = self.registers[*source].clone();
2333                    pc += 1;
2334                }
2335                OpCode::CopyRegisterIfNull { source, dest } => {
2336                    if self.registers[*dest].is_null() {
2337                        self.registers[*dest] = self.registers[*source].clone();
2338                    }
2339                    pc += 1;
2340                }
2341                OpCode::GetEventType { dest } => {
2342                    self.registers[*dest] = json!(event_type);
2343                    pc += 1;
2344                }
2345                OpCode::CreateObject { dest } => {
2346                    self.registers[*dest] = json!({});
2347                    pc += 1;
2348                }
2349                OpCode::SetField {
2350                    object,
2351                    path,
2352                    value,
2353                } => {
2354                    let old_value = self
2355                        .is_debug_enabled()
2356                        .then(|| Self::get_value_at_path(&self.registers[*object], path))
2357                        .flatten();
2358                    self.set_field_auto_vivify(*object, path, *value)?;
2359                    if should_emit(path) {
2360                        dirty_tracker.mark_replaced(path);
2361                    }
2362                    if self.is_debug_enabled() {
2363                        self.emit_debug(|| VmDebugEvent::FieldWrite {
2364                            entity_name: entity_name.to_string(),
2365                            event_type: event_type.to_string(),
2366                            op: "set_field".to_string(),
2367                            path: path.clone(),
2368                            old_value,
2369                            new_value: Self::get_value_at_path(&self.registers[*object], path),
2370                            applied: true,
2371                            reason: None,
2372                        });
2373                    }
2374                    pc += 1;
2375                }
2376                OpCode::SetFields { object, fields } => {
2377                    for (path, value_reg) in fields {
2378                        let old_value = self
2379                            .is_debug_enabled()
2380                            .then(|| Self::get_value_at_path(&self.registers[*object], path))
2381                            .flatten();
2382                        self.set_field_auto_vivify(*object, path, *value_reg)?;
2383                        if should_emit(path) {
2384                            dirty_tracker.mark_replaced(path);
2385                        }
2386                        if self.is_debug_enabled() {
2387                            self.emit_debug(|| VmDebugEvent::FieldWrite {
2388                                entity_name: entity_name.to_string(),
2389                                event_type: event_type.to_string(),
2390                                op: "set_fields".to_string(),
2391                                path: path.clone(),
2392                                old_value,
2393                                new_value: Self::get_value_at_path(&self.registers[*object], path),
2394                                applied: true,
2395                                reason: None,
2396                            });
2397                        }
2398                    }
2399                    pc += 1;
2400                }
2401                OpCode::GetField { object, path, dest } => {
2402                    let value = self.get_field(*object, path)?;
2403                    self.registers[*dest] = value;
2404                    pc += 1;
2405                }
2406                OpCode::AbortIfNullKey {
2407                    key,
2408                    is_account_event,
2409                } => {
2410                    let key_value = &self.registers[*key];
2411                    if key_value.is_null() && *is_account_event {
2412                        self.emit_debug(|| VmDebugEvent::ReadOrInitState {
2413                            entity_name: entity_name.to_string(),
2414                            event_type: event_type.to_string(),
2415                            key: key_value.clone(),
2416                            existing_state: None,
2417                            loaded_state: Value::Null,
2418                            skipped_reason: Some("null_primary_key".to_string()),
2419                        });
2420                        tracing::debug!(
2421                            event_type = %event_type,
2422                            "AbortIfNullKey: key is null for account state event, \
2423                             returning empty mutations for queueing"
2424                        );
2425                        return Ok(Vec::new());
2426                    }
2427
2428                    pc += 1;
2429                }
2430                OpCode::ReadOrInitState {
2431                    state_id: _,
2432                    key,
2433                    default,
2434                    dest,
2435                } => {
2436                    let actual_state_id = override_state_id;
2437                    let entity_name_owned = entity_name.to_string();
2438                    self.states
2439                        .entry(actual_state_id)
2440                        .or_insert_with(|| StateTable {
2441                            data: DashMap::new(),
2442                            access_times: DashMap::new(),
2443                            lookup_indexes: HashMap::new(),
2444                            temporal_indexes: HashMap::new(),
2445                            pda_reverse_lookups: HashMap::new(),
2446                            pending_updates: DashMap::new(),
2447                            pending_instruction_events: DashMap::new(),
2448                            last_account_data: DashMap::new(),
2449                            version_tracker: VersionTracker::new(),
2450                            instruction_dedup_cache: VersionTracker::with_capacity(
2451                                DEFAULT_MAX_INSTRUCTION_DEDUP_ENTRIES,
2452                            ),
2453                            config: StateTableConfig::default(),
2454                            entity_name: entity_name_owned,
2455                            recent_tx_instructions: std::sync::Mutex::new(LruCache::new(
2456                                NonZeroUsize::new(1000).unwrap(),
2457                            )),
2458                            deferred_when_ops: DashMap::new(),
2459                        });
2460                    let key_value = self.registers[*key].clone();
2461                    // Warn if key is null for account state events (not instruction events or CPI events)
2462                    let warn_null_key = key_value.is_null()
2463                        && event_type.ends_with("State")
2464                        && !event_type.ends_with("IxState")
2465                        && !event_type.ends_with("CpiEvent");
2466
2467                    if warn_null_key {
2468                        self.add_warning(format!(
2469                            "ReadOrInitState: key register {} is NULL for account state, event_type={}",
2470                            key, event_type
2471                        ));
2472                    }
2473
2474                    let state = self
2475                        .states
2476                        .get(&actual_state_id)
2477                        .ok_or("State table not found")?;
2478
2479                    if !key_value.is_null() {
2480                        if let Some(ctx) = &self.current_context {
2481                            // Account updates: use recency check to discard stale updates
2482                            // IMPORTANT: Use account address (PDA) as the key, not entity key,
2483                            // because multiple accounts can update the same entity and each
2484                            // has its own independent write_version
2485                            if ctx.is_account_update() {
2486                                if let (Some(slot), Some(write_version)) =
2487                                    (ctx.slot, ctx.write_version)
2488                                {
2489                                    // Get account address from event_value for proper tracking
2490                                    let account_address = event_value
2491                                        .get("__account_address")
2492                                        .cloned()
2493                                        .unwrap_or_else(|| key_value.clone());
2494
2495                                    if !state.is_fresh_update(
2496                                        &account_address,
2497                                        event_type,
2498                                        slot,
2499                                        write_version,
2500                                    ) {
2501                                        self.emit_debug(|| VmDebugEvent::ReadOrInitState {
2502                                            entity_name: entity_name.to_string(),
2503                                            event_type: event_type.to_string(),
2504                                            key: key_value.clone(),
2505                                            existing_state: None,
2506                                            loaded_state: Value::Null,
2507                                            skipped_reason: Some(
2508                                                "stale_account_update".to_string(),
2509                                            ),
2510                                        });
2511                                        self.add_warning(format!(
2512                                            "Stale account update skipped: slot={}, write_version={}, account={}",
2513                                            slot, write_version, account_address
2514                                        ));
2515                                        return Ok(Vec::new());
2516                                    }
2517                                }
2518                            }
2519                            // Instruction updates: process all, but skip exact duplicates
2520                            else if ctx.is_instruction_update() {
2521                                if let (Some(slot), Some(txn_index)) = (ctx.slot, ctx.txn_index) {
2522                                    if state.is_duplicate_instruction(
2523                                        &key_value, event_type, slot, txn_index,
2524                                    ) {
2525                                        self.emit_debug(|| VmDebugEvent::ReadOrInitState {
2526                                            entity_name: entity_name.to_string(),
2527                                            event_type: event_type.to_string(),
2528                                            key: key_value.clone(),
2529                                            existing_state: None,
2530                                            loaded_state: Value::Null,
2531                                            skipped_reason: Some(
2532                                                "duplicate_instruction".to_string(),
2533                                            ),
2534                                        });
2535                                        self.add_warning(format!(
2536                                            "Duplicate instruction skipped: slot={}, txn_index={}",
2537                                            slot, txn_index
2538                                        ));
2539                                        return Ok(Vec::new());
2540                                    }
2541                                }
2542                            }
2543                        }
2544                    }
2545                    let existing_state = state.get_and_touch(&key_value);
2546                    let value = existing_state.clone().unwrap_or_else(|| default.clone());
2547
2548                    self.emit_debug(|| VmDebugEvent::ReadOrInitState {
2549                        entity_name: entity_name.to_string(),
2550                        event_type: event_type.to_string(),
2551                        key: key_value,
2552                        existing_state,
2553                        loaded_state: value.clone(),
2554                        skipped_reason: None,
2555                    });
2556
2557                    self.registers[*dest] = value;
2558                    pc += 1;
2559                }
2560                OpCode::UpdateState {
2561                    state_id: _,
2562                    key,
2563                    value,
2564                } => {
2565                    let actual_state_id = override_state_id;
2566                    let state = self
2567                        .states
2568                        .get(&actual_state_id)
2569                        .ok_or("State table not found")?;
2570                    let key_value = self.registers[*key].clone();
2571                    let value_data = self.registers[*value].clone();
2572
2573                    state.insert_with_eviction(key_value, value_data);
2574                    pc += 1;
2575                }
2576                OpCode::AppendToArray {
2577                    object,
2578                    path,
2579                    value,
2580                } => {
2581                    let appended_value = self.registers[*value].clone();
2582                    let max_len = self
2583                        .states
2584                        .get(&override_state_id)
2585                        .map(|s| s.max_array_length())
2586                        .unwrap_or(DEFAULT_MAX_ARRAY_LENGTH);
2587                    self.append_to_array(*object, path, *value, max_len)?;
2588                    if should_emit(path) {
2589                        dirty_tracker.mark_appended(path, appended_value);
2590                    }
2591                    pc += 1;
2592                }
2593                OpCode::GetCurrentTimestamp { dest } => {
2594                    let timestamp = std::time::SystemTime::now()
2595                        .duration_since(std::time::UNIX_EPOCH)
2596                        .unwrap()
2597                        .as_secs() as i64;
2598                    self.registers[*dest] = json!(timestamp);
2599                    pc += 1;
2600                }
2601                OpCode::CreateEvent { dest, event_value } => {
2602                    let timestamp = std::time::SystemTime::now()
2603                        .duration_since(std::time::UNIX_EPOCH)
2604                        .unwrap()
2605                        .as_secs() as i64;
2606
2607                    // Filter out __update_context from the event data
2608                    let mut event_data = self.registers[*event_value].clone();
2609                    if let Some(obj) = event_data.as_object_mut() {
2610                        obj.remove("__update_context");
2611                    }
2612
2613                    // Create event with timestamp, data, and optional slot/signature from context
2614                    let mut event = serde_json::Map::new();
2615                    event.insert("timestamp".to_string(), json!(timestamp));
2616                    event.insert("data".to_string(), event_data);
2617
2618                    // Add slot and signature if available from current context
2619                    if let Some(ref ctx) = self.current_context {
2620                        if let Some(slot) = ctx.slot {
2621                            event.insert("slot".to_string(), json!(slot));
2622                        }
2623                        if let Some(ref signature) = ctx.signature {
2624                            event.insert("signature".to_string(), json!(signature));
2625                        }
2626                    }
2627
2628                    self.registers[*dest] = Value::Object(event);
2629                    pc += 1;
2630                }
2631                OpCode::CreateCapture {
2632                    dest,
2633                    capture_value,
2634                } => {
2635                    let timestamp = std::time::SystemTime::now()
2636                        .duration_since(std::time::UNIX_EPOCH)
2637                        .unwrap()
2638                        .as_secs() as i64;
2639
2640                    // Get the capture data (already filtered by load_field)
2641                    let capture_data = self.registers[*capture_value].clone();
2642
2643                    // Extract account_address from the original event if available
2644                    let account_address = event_value
2645                        .get("__account_address")
2646                        .and_then(|v| v.as_str())
2647                        .unwrap_or("")
2648                        .to_string();
2649
2650                    // Create capture wrapper with timestamp, account_address, data, and optional slot/signature
2651                    let mut capture = serde_json::Map::new();
2652                    capture.insert("timestamp".to_string(), json!(timestamp));
2653                    capture.insert("account_address".to_string(), json!(account_address));
2654                    capture.insert("data".to_string(), capture_data);
2655
2656                    // Add slot and signature if available from current context
2657                    if let Some(ref ctx) = self.current_context {
2658                        if let Some(slot) = ctx.slot {
2659                            capture.insert("slot".to_string(), json!(slot));
2660                        }
2661                        if let Some(ref signature) = ctx.signature {
2662                            capture.insert("signature".to_string(), json!(signature));
2663                        }
2664                    }
2665
2666                    self.registers[*dest] = Value::Object(capture);
2667                    pc += 1;
2668                }
2669                OpCode::Transform {
2670                    source,
2671                    dest,
2672                    transformation,
2673                } => {
2674                    if source == dest {
2675                        self.transform_in_place(*source, transformation)?;
2676                    } else {
2677                        let source_value = &self.registers[*source];
2678                        let value = Self::apply_transformation(source_value, transformation)?;
2679                        self.registers[*dest] = value;
2680                    }
2681                    pc += 1;
2682                }
2683                OpCode::EmitMutation {
2684                    entity_name,
2685                    key,
2686                    state,
2687                } => {
2688                    let primary_key = self.registers[*key].clone();
2689                    let dirty_fields: Vec<String> =
2690                        dirty_tracker.dirty_paths().into_iter().collect();
2691
2692                    if primary_key.is_null() || dirty_tracker.is_empty() {
2693                        let reason = if dirty_tracker.is_empty() {
2694                            "no_fields_modified"
2695                        } else {
2696                            "null_primary_key"
2697                        };
2698                        self.emit_debug(|| VmDebugEvent::EmitMutation {
2699                            entity_name: entity_name.clone(),
2700                            event_type: event_type.to_string(),
2701                            key: primary_key.clone(),
2702                            emitted: false,
2703                            reason: Some(reason.to_string()),
2704                            patch: None,
2705                            dirty_fields,
2706                        });
2707                        self.add_warning(format!(
2708                            "Skipping mutation for entity '{}': {} (dirty_fields={})",
2709                            entity_name,
2710                            reason,
2711                            dirty_tracker.len()
2712                        ));
2713                    } else {
2714                        let patch =
2715                            self.extract_partial_state_with_tracker(*state, &dirty_tracker)?;
2716
2717                        let append = dirty_tracker.appended_paths();
2718                        let mutation = Mutation {
2719                            export: entity_name.clone(),
2720                            key: primary_key,
2721                            patch,
2722                            append,
2723                        };
2724                        self.emit_debug(|| VmDebugEvent::EmitMutation {
2725                            entity_name: entity_name.clone(),
2726                            event_type: event_type.to_string(),
2727                            key: mutation.key.clone(),
2728                            emitted: true,
2729                            reason: None,
2730                            patch: Some(mutation.patch.clone()),
2731                            dirty_fields,
2732                        });
2733                        output.push(mutation);
2734                    }
2735                    pc += 1;
2736                }
2737                OpCode::SetFieldIfNull {
2738                    object,
2739                    path,
2740                    value,
2741                } => {
2742                    let old_value = self
2743                        .is_debug_enabled()
2744                        .then(|| Self::get_value_at_path(&self.registers[*object], path))
2745                        .flatten();
2746                    let was_set = self.set_field_if_null(*object, path, *value)?;
2747                    if was_set && should_emit(path) {
2748                        dirty_tracker.mark_replaced(path);
2749                    }
2750                    if self.is_debug_enabled() {
2751                        self.emit_debug(|| VmDebugEvent::FieldWrite {
2752                            entity_name: entity_name.to_string(),
2753                            event_type: event_type.to_string(),
2754                            op: "set_field_if_null".to_string(),
2755                            path: path.clone(),
2756                            old_value,
2757                            new_value: Self::get_value_at_path(&self.registers[*object], path),
2758                            applied: was_set,
2759                            reason: (!was_set).then_some("already_set".to_string()),
2760                        });
2761                    }
2762                    pc += 1;
2763                }
2764                OpCode::SetFieldMax {
2765                    object,
2766                    path,
2767                    value,
2768                } => {
2769                    let was_updated = self.set_field_max(*object, path, *value)?;
2770                    if was_updated && should_emit(path) {
2771                        dirty_tracker.mark_replaced(path);
2772                    }
2773                    pc += 1;
2774                }
2775                OpCode::UpdateTemporalIndex {
2776                    state_id: _,
2777                    index_name,
2778                    lookup_value,
2779                    primary_key,
2780                    timestamp,
2781                } => {
2782                    let actual_state_id = override_state_id;
2783                    let state = self
2784                        .states
2785                        .get_mut(&actual_state_id)
2786                        .ok_or("State table not found")?;
2787                    let index = state
2788                        .temporal_indexes
2789                        .entry(index_name.clone())
2790                        .or_insert_with(TemporalIndex::new);
2791
2792                    let lookup_val = self.registers[*lookup_value].clone();
2793                    let pk_val = self.registers[*primary_key].clone();
2794                    let ts_val = if let Some(val) = self.registers[*timestamp].as_i64() {
2795                        val
2796                    } else if let Some(val) = self.registers[*timestamp].as_u64() {
2797                        val as i64
2798                    } else {
2799                        return Err(format!(
2800                            "Timestamp must be a number (i64 or u64), got: {:?}",
2801                            self.registers[*timestamp]
2802                        )
2803                        .into());
2804                    };
2805
2806                    index.insert(lookup_val, pk_val, ts_val);
2807                    pc += 1;
2808                }
2809                OpCode::LookupTemporalIndex {
2810                    state_id: _,
2811                    index_name,
2812                    lookup_value,
2813                    timestamp,
2814                    dest,
2815                } => {
2816                    let actual_state_id = override_state_id;
2817                    let state = self
2818                        .states
2819                        .get(&actual_state_id)
2820                        .ok_or("State table not found")?;
2821                    let lookup_val = &self.registers[*lookup_value];
2822
2823                    let result = if self.registers[*timestamp].is_null() {
2824                        if let Some(index) = state.temporal_indexes.get(index_name) {
2825                            index.lookup_latest(lookup_val).unwrap_or(Value::Null)
2826                        } else {
2827                            Value::Null
2828                        }
2829                    } else {
2830                        let ts_val = if let Some(val) = self.registers[*timestamp].as_i64() {
2831                            val
2832                        } else if let Some(val) = self.registers[*timestamp].as_u64() {
2833                            val as i64
2834                        } else {
2835                            return Err(format!(
2836                                "Timestamp must be a number (i64 or u64), got: {:?}",
2837                                self.registers[*timestamp]
2838                            )
2839                            .into());
2840                        };
2841
2842                        if let Some(index) = state.temporal_indexes.get(index_name) {
2843                            index.lookup(lookup_val, ts_val).unwrap_or(Value::Null)
2844                        } else {
2845                            Value::Null
2846                        }
2847                    };
2848
2849                    self.registers[*dest] = result;
2850                    pc += 1;
2851                }
2852                OpCode::UpdateLookupIndex {
2853                    state_id: _,
2854                    index_name,
2855                    lookup_value,
2856                    primary_key,
2857                } => {
2858                    let actual_state_id = override_state_id;
2859                    let state = self
2860                        .states
2861                        .get_mut(&actual_state_id)
2862                        .ok_or("State table not found")?;
2863                    let index = state
2864                        .lookup_indexes
2865                        .entry(index_name.clone())
2866                        .or_insert_with(LookupIndex::new);
2867
2868                    let lookup_val = self.registers[*lookup_value].clone();
2869                    let pk_val = self.registers[*primary_key].clone();
2870
2871                    index.insert(lookup_val.clone(), pk_val);
2872
2873                    // Track lookup keys so process_event can flush queued account updates
2874                    if let Some(key_str) = lookup_val.as_str() {
2875                        self.last_lookup_index_keys.push(key_str.to_string());
2876                    }
2877
2878                    pc += 1;
2879                }
2880                OpCode::LookupIndex {
2881                    state_id: _,
2882                    index_name,
2883                    lookup_value,
2884                    dest,
2885                } => {
2886                    let actual_state_id = override_state_id;
2887                    let mut current_value = self.registers[*lookup_value].clone();
2888                    let original_lookup_value = current_value.clone();
2889                    let mut hops = if self.is_debug_enabled() {
2890                        Some(Vec::<VmLookupHop>::new())
2891                    } else {
2892                        None
2893                    };
2894                    let mut miss_kind: Option<String> = None;
2895
2896                    const MAX_CHAIN_DEPTH: usize = 5;
2897                    let mut iterations = 0;
2898
2899                    let final_result = if self.states.contains_key(&actual_state_id) {
2900                        loop {
2901                            iterations += 1;
2902                            if iterations > MAX_CHAIN_DEPTH {
2903                                break current_value;
2904                            }
2905
2906                            let resolved = self
2907                                .states
2908                                .get(&actual_state_id)
2909                                .and_then(|state| {
2910                                    if let Some(index) = state.lookup_indexes.get(index_name) {
2911                                        if let Some(found) = index.lookup(&current_value) {
2912                                            return Some(found);
2913                                        }
2914                                    }
2915
2916                                    for (name, index) in state.lookup_indexes.iter() {
2917                                        if name == index_name {
2918                                            continue;
2919                                        }
2920                                        if let Some(found) = index.lookup(&current_value) {
2921                                            return Some(found);
2922                                        }
2923                                    }
2924
2925                                    None
2926                                })
2927                                .unwrap_or(Value::Null);
2928
2929                            let lookup_result = resolved.clone();
2930                            if let Some(hops) = hops.as_mut() {
2931                                hops.push(VmLookupHop {
2932                                    source: "lookup_index".to_string(),
2933                                    input: current_value.clone(),
2934                                    result: lookup_result.clone(),
2935                                    chained: false,
2936                                });
2937                            }
2938
2939                            let mut resolved_from_pda = false;
2940                            let resolved = if resolved.is_null() {
2941                                if let Some(pda_str) = current_value.as_str() {
2942                                    resolved_from_pda = true;
2943                                    let pda_result = self
2944                                        .states
2945                                        .get_mut(&actual_state_id)
2946                                        .and_then(|state_mut| {
2947                                            state_mut
2948                                                .pda_reverse_lookups
2949                                                .get_mut("default_pda_lookup")
2950                                        })
2951                                        .and_then(|pda_lookup| pda_lookup.lookup(pda_str))
2952                                        .map(Value::String)
2953                                        .unwrap_or(Value::Null);
2954                                    if let Some(hops) = hops.as_mut() {
2955                                        hops.push(VmLookupHop {
2956                                            source: "pda_reverse_lookup".to_string(),
2957                                            input: current_value.clone(),
2958                                            result: pda_result.clone(),
2959                                            chained: false,
2960                                        });
2961                                    }
2962                                    pda_result
2963                                } else {
2964                                    Value::Null
2965                                }
2966                            } else {
2967                                resolved
2968                            };
2969
2970                            if resolved.is_null() {
2971                                if iterations == 1 {
2972                                    if let Some(pda_str) = current_value.as_str() {
2973                                        self.last_pda_lookup_miss = Some(pda_str.to_string());
2974                                        miss_kind = Some("pda_lookup_miss".to_string());
2975                                    } else {
2976                                        miss_kind = Some("lookup_index_miss".to_string());
2977                                    }
2978                                }
2979                                break Value::Null;
2980                            }
2981
2982                            let can_chain =
2983                                self.can_resolve_further(&resolved, actual_state_id, index_name);
2984
2985                            if !can_chain {
2986                                if resolved_from_pda {
2987                                    if let Some(resolved_str) = resolved.as_str() {
2988                                        self.last_lookup_index_miss =
2989                                            Some(resolved_str.to_string());
2990                                    }
2991                                    miss_kind = Some("lookup_chain_incomplete".to_string());
2992                                    break Value::Null;
2993                                }
2994                                break resolved;
2995                            }
2996
2997                            if let Some(hops) = hops.as_mut() {
2998                                if let Some(last) = hops.last_mut() {
2999                                    last.chained = true;
3000                                }
3001                            }
3002
3003                            current_value = resolved;
3004                        }
3005                    } else {
3006                        miss_kind = Some("state_not_initialized".to_string());
3007                        Value::Null
3008                    };
3009
3010                    self.emit_debug(|| VmDebugEvent::LookupIndex {
3011                        entity_name: entity_name.to_string(),
3012                        event_type: event_type.to_string(),
3013                        index_name: index_name.clone(),
3014                        lookup_value: original_lookup_value,
3015                        hops: hops.unwrap_or_default(),
3016                        final_result: final_result.clone(),
3017                        miss_kind,
3018                    });
3019
3020                    self.registers[*dest] = final_result;
3021                    pc += 1;
3022                }
3023                OpCode::SetFieldSum {
3024                    object,
3025                    path,
3026                    value,
3027                } => {
3028                    let was_updated = self.set_field_sum(*object, path, *value)?;
3029                    if was_updated && should_emit(path) {
3030                        dirty_tracker.mark_replaced(path);
3031                    }
3032                    pc += 1;
3033                }
3034                OpCode::SetFieldIncrement { object, path } => {
3035                    let was_updated = self.set_field_increment(*object, path)?;
3036                    if was_updated && should_emit(path) {
3037                        dirty_tracker.mark_replaced(path);
3038                    }
3039                    pc += 1;
3040                }
3041                OpCode::SetFieldMin {
3042                    object,
3043                    path,
3044                    value,
3045                } => {
3046                    let was_updated = self.set_field_min(*object, path, *value)?;
3047                    if was_updated && should_emit(path) {
3048                        dirty_tracker.mark_replaced(path);
3049                    }
3050                    pc += 1;
3051                }
3052                OpCode::AddToUniqueSet {
3053                    state_id: _,
3054                    set_name,
3055                    value,
3056                    count_object,
3057                    count_path,
3058                } => {
3059                    let value_to_add = self.registers[*value].clone();
3060
3061                    // Store the unique set within the entity object, not in the state table
3062                    // This ensures each entity instance has its own unique set
3063                    let set_field_path = format!("__unique_set:{}", set_name);
3064
3065                    // Get or create the unique set from the entity object
3066                    let mut set: HashSet<Value> =
3067                        if let Ok(existing) = self.get_field(*count_object, &set_field_path) {
3068                            if !existing.is_null() {
3069                                serde_json::from_value(existing).unwrap_or_default()
3070                            } else {
3071                                HashSet::new()
3072                            }
3073                        } else {
3074                            HashSet::new()
3075                        };
3076
3077                    // Add value to set
3078                    let was_new = set.insert(value_to_add);
3079
3080                    // Store updated set back in the entity object
3081                    let set_as_vec: Vec<Value> = set.iter().cloned().collect();
3082                    self.registers[100] = serde_json::to_value(set_as_vec)?;
3083                    self.set_field_auto_vivify(*count_object, &set_field_path, 100)?;
3084
3085                    // Update the count field in the object
3086                    if was_new {
3087                        self.registers[100] = Value::Number(serde_json::Number::from(set.len()));
3088                        self.set_field_auto_vivify(*count_object, count_path, 100)?;
3089                        if should_emit(count_path) {
3090                            dirty_tracker.mark_replaced(count_path);
3091                        }
3092                    }
3093
3094                    pc += 1;
3095                }
3096                OpCode::ConditionalSetField {
3097                    object,
3098                    path,
3099                    value,
3100                    condition_field,
3101                    condition_op,
3102                    condition_value,
3103                } => {
3104                    let field_value = self.load_field(event_value, condition_field, None)?;
3105                    let condition_met =
3106                        self.evaluate_comparison(&field_value, condition_op, condition_value)?;
3107
3108                    if condition_met {
3109                        self.set_field_auto_vivify(*object, path, *value)?;
3110                        if should_emit(path) {
3111                            dirty_tracker.mark_replaced(path);
3112                        }
3113                    }
3114                    pc += 1;
3115                }
3116                OpCode::SetFieldWhen {
3117                    object,
3118                    path,
3119                    value,
3120                    when_instruction,
3121                    entity_name,
3122                    key_reg,
3123                    condition_field,
3124                    condition_op,
3125                    condition_value,
3126                } => {
3127                    let actual_state_id = override_state_id;
3128                    let condition_met = if let (Some(field), Some(op), Some(cond_value)) = (
3129                        condition_field.as_ref(),
3130                        condition_op.as_ref(),
3131                        condition_value.as_ref(),
3132                    ) {
3133                        let field_value = self.load_field(event_value, field, None)?;
3134                        self.evaluate_comparison(&field_value, op, cond_value)?
3135                    } else {
3136                        true
3137                    };
3138
3139                    if !condition_met {
3140                        pc += 1;
3141                        continue;
3142                    }
3143
3144                    let signature = self
3145                        .current_context
3146                        .as_ref()
3147                        .and_then(|c| c.signature.clone())
3148                        .unwrap_or_default();
3149
3150                    let emit = should_emit(path);
3151
3152                    let instruction_seen = if !signature.is_empty() {
3153                        if let Some(state) = self.states.get(&actual_state_id) {
3154                            let mut cache = state.recent_tx_instructions.lock().unwrap();
3155                            cache
3156                                .get(&signature)
3157                                .map(|set| set.contains(when_instruction))
3158                                .unwrap_or(false)
3159                        } else {
3160                            false
3161                        }
3162                    } else {
3163                        false
3164                    };
3165
3166                    if instruction_seen {
3167                        self.set_field_auto_vivify(*object, path, *value)?;
3168                        if emit {
3169                            dirty_tracker.mark_replaced(path);
3170                        }
3171                    } else if !signature.is_empty() {
3172                        let deferred = DeferredWhenOperation {
3173                            entity_name: entity_name.clone(),
3174                            primary_key: self.registers[*key_reg].clone(),
3175                            field_path: path.clone(),
3176                            field_value: self.registers[*value].clone(),
3177                            when_instruction: when_instruction.clone(),
3178                            signature: signature.clone(),
3179                            slot: self
3180                                .current_context
3181                                .as_ref()
3182                                .and_then(|c| c.slot)
3183                                .unwrap_or(0),
3184                            deferred_at: std::time::SystemTime::now()
3185                                .duration_since(std::time::UNIX_EPOCH)
3186                                .unwrap()
3187                                .as_secs() as i64,
3188                            emit,
3189                        };
3190
3191                        if let Some(state) = self.states.get(&actual_state_id) {
3192                            let key = (signature, when_instruction.clone());
3193                            state
3194                                .deferred_when_ops
3195                                .entry(key)
3196                                .or_insert_with(Vec::new)
3197                                .push(deferred);
3198                        }
3199                    }
3200
3201                    pc += 1;
3202                }
3203                OpCode::SetFieldUnlessStopped {
3204                    object,
3205                    path,
3206                    value,
3207                    stop_field,
3208                    stop_instruction,
3209                    entity_name,
3210                    key_reg: _,
3211                } => {
3212                    let stop_value = self.get_field(*object, stop_field).unwrap_or(Value::Null);
3213                    let stopped = stop_value.as_bool().unwrap_or(false);
3214                    let old_value = self
3215                        .is_debug_enabled()
3216                        .then(|| Self::get_value_at_path(&self.registers[*object], path))
3217                        .flatten();
3218
3219                    if stopped {
3220                        self.emit_debug(|| VmDebugEvent::FieldWrite {
3221                            entity_name: entity_name.clone(),
3222                            event_type: event_type.to_string(),
3223                            op: "set_field_unless_stopped".to_string(),
3224                            path: path.clone(),
3225                            old_value,
3226                            new_value: Self::get_value_at_path(&self.registers[*object], path),
3227                            applied: false,
3228                            reason: Some(format!("stop_flag:{}", stop_instruction)),
3229                        });
3230                        tracing::debug!(
3231                            entity = %entity_name,
3232                            field = %path,
3233                            stop_field = %stop_field,
3234                            stop_instruction = %stop_instruction,
3235                            "stop flag set; skipping field update"
3236                        );
3237                        pc += 1;
3238                        continue;
3239                    }
3240
3241                    self.set_field_auto_vivify(*object, path, *value)?;
3242                    if should_emit(path) {
3243                        dirty_tracker.mark_replaced(path);
3244                    }
3245                    if self.is_debug_enabled() {
3246                        self.emit_debug(|| VmDebugEvent::FieldWrite {
3247                            entity_name: entity_name.clone(),
3248                            event_type: event_type.to_string(),
3249                            op: "set_field_unless_stopped".to_string(),
3250                            path: path.clone(),
3251                            old_value,
3252                            new_value: Self::get_value_at_path(&self.registers[*object], path),
3253                            applied: true,
3254                            reason: None,
3255                        });
3256                    }
3257                    pc += 1;
3258                }
3259                OpCode::ConditionalIncrement {
3260                    object,
3261                    path,
3262                    condition_field,
3263                    condition_op,
3264                    condition_value,
3265                } => {
3266                    let field_value = self.load_field(event_value, condition_field, None)?;
3267                    let condition_met =
3268                        self.evaluate_comparison(&field_value, condition_op, condition_value)?;
3269
3270                    if condition_met {
3271                        let was_updated = self.set_field_increment(*object, path)?;
3272                        if was_updated && should_emit(path) {
3273                            dirty_tracker.mark_replaced(path);
3274                        }
3275                    }
3276                    pc += 1;
3277                }
3278                OpCode::EvaluateComputedFields {
3279                    state,
3280                    computed_paths,
3281                } => {
3282                    if let Some(evaluator) = entity_evaluator {
3283                        let old_values: Vec<_> = computed_paths
3284                            .iter()
3285                            .map(|path| Self::get_value_at_path(&self.registers[*state], path))
3286                            .collect();
3287
3288                        let state_value = &mut self.registers[*state];
3289                        let context_slot = self.current_context.as_ref().and_then(|c| c.slot);
3290                        let context_timestamp = self
3291                            .current_context
3292                            .as_ref()
3293                            .map(|c| c.timestamp())
3294                            .unwrap_or_else(|| {
3295                                std::time::SystemTime::now()
3296                                    .duration_since(std::time::UNIX_EPOCH)
3297                                    .unwrap()
3298                                    .as_secs() as i64
3299                            });
3300                        let eval_result = evaluator(state_value, context_slot, context_timestamp);
3301
3302                        if eval_result.is_ok() {
3303                            for (path, old_value) in computed_paths.iter().zip(old_values.iter()) {
3304                                let new_value =
3305                                    Self::get_value_at_path(&self.registers[*state], path);
3306
3307                                if new_value != *old_value && should_emit(path) {
3308                                    dirty_tracker.mark_replaced(path);
3309                                }
3310                            }
3311                        }
3312                    }
3313                    pc += 1;
3314                }
3315                OpCode::QueueResolver {
3316                    state_id: _,
3317                    entity_name,
3318                    resolver,
3319                    input_path,
3320                    input_value,
3321                    url_template,
3322                    strategy,
3323                    extracts,
3324                    condition,
3325                    schedule_at,
3326                    state,
3327                    key,
3328                } => {
3329                    let actual_state_id = override_state_id;
3330
3331                    // Reprocessed account data may safely register a future callback, but
3332                    // must not resolve immediately and lock stale data in via SetOnce.
3333                    let skip_immediate_resolver = self
3334                        .current_context
3335                        .as_ref()
3336                        .map(|c| c.skip_resolvers)
3337                        .unwrap_or(false);
3338
3339                    // Evaluate condition if present
3340                    if let Some(cond) = condition {
3341                        let field_val =
3342                            Self::get_value_at_path(&self.registers[*state], &cond.field_path)
3343                                .unwrap_or(Value::Null);
3344                        let condition_met =
3345                            self.evaluate_comparison(&field_val, &cond.op, &cond.value)?;
3346
3347                        if !condition_met {
3348                            pc += 1;
3349                            continue;
3350                        }
3351                    }
3352
3353                    // Check schedule_at: defer if target slot is in the future
3354                    if let Some(schedule_path) = schedule_at {
3355                        let target_val =
3356                            Self::get_value_at_path(&self.registers[*state], schedule_path);
3357
3358                        match target_val.as_ref().and_then(|value| {
3359                            value.as_u64().or_else(|| {
3360                                value.as_str().and_then(|text| text.parse::<u64>().ok())
3361                            })
3362                        }) {
3363                            Some(target_slot) => {
3364                                let current_slot = self
3365                                    .current_context
3366                                    .as_ref()
3367                                    .and_then(|ctx| ctx.slot)
3368                                    .unwrap_or(0);
3369                                if current_slot < target_slot {
3370                                    let key_value = &self.registers[*key];
3371                                    if !key_value.is_null() {
3372                                        self.scheduled_callbacks.push((
3373                                            target_slot,
3374                                            ScheduledCallback {
3375                                                state_id: actual_state_id,
3376                                                entity_name: entity_name.clone(),
3377                                                primary_key: key_value.clone(),
3378                                                resolver: resolver.clone(),
3379                                                url_template: url_template.clone(),
3380                                                input_value: input_value.clone(),
3381                                                input_path: input_path.clone(),
3382                                                condition: condition.clone(),
3383                                                strategy: strategy.clone(),
3384                                                extracts: extracts.clone(),
3385                                                retry_count: 0,
3386                                            },
3387                                        ));
3388                                    }
3389                                    pc += 1;
3390                                    continue;
3391                                }
3392                                // current_slot >= target_slot: fall through to immediate resolution
3393                            }
3394                            None => {
3395                                // schedule_at path is missing or value is not a u64 —
3396                                // skip resolver entirely rather than executing immediately,
3397                                // since the state likely hasn't been fully populated yet.
3398                                pc += 1;
3399                                continue;
3400                            }
3401                        }
3402                    }
3403
3404                    if skip_immediate_resolver {
3405                        pc += 1;
3406                        continue;
3407                    }
3408
3409                    // Build resolver input from template, literal value, or field path
3410                    let resolved_input = if let Some(template) = url_template {
3411                        crate::scheduler::build_url_from_template(template, &self.registers[*state])
3412                            .map(Value::String)
3413                    } else if let Some(value) = input_value {
3414                        Some(value.clone())
3415                    } else if let Some(path) = input_path.as_ref() {
3416                        Self::get_value_at_path(&self.registers[*state], path)
3417                    } else {
3418                        None
3419                    };
3420
3421                    if let Some(input) = resolved_input {
3422                        let key_value = &self.registers[*key];
3423
3424                        if input.is_null() || key_value.is_null() {
3425                            pc += 1;
3426                            continue;
3427                        }
3428
3429                        if matches!(strategy, ResolveStrategy::SetOnce)
3430                            // all(): fire resolver if any target is still null.
3431                            // Already-set fields are protected from overwrite by
3432                            // set_value_at_path's SetOnce null-source guard.
3433                            && extracts.iter().all(|extract| {
3434                                match Self::get_value_at_path(
3435                                    &self.registers[*state],
3436                                    &extract.target_path,
3437                                ) {
3438                                    Some(value) => !value.is_null(),
3439                                    None => false,
3440                                }
3441                            })
3442                        {
3443                            pc += 1;
3444                            continue;
3445                        }
3446
3447                        let cache_key = resolver_cache_key(resolver, &input);
3448
3449                        match self.get_cached_resolver_value(&cache_key) {
3450                            Some(CachedResolverValue::Resolved(cached)) => {
3451                                Self::apply_resolver_extractions_to_value(
3452                                    &mut self.registers[*state],
3453                                    &cached,
3454                                    extracts,
3455                                    &mut dirty_tracker,
3456                                    &should_emit,
3457                                )?;
3458                            }
3459                            Some(CachedResolverValue::Negative) => {}
3460                            None => {
3461                                let target = ResolverTarget {
3462                                    state_id: actual_state_id,
3463                                    entity_name: entity_name.clone(),
3464                                    primary_key: self.registers[*key].clone(),
3465                                    extracts: extracts.clone(),
3466                                };
3467
3468                                self.enqueue_resolver_request(
3469                                    cache_key,
3470                                    resolver.clone(),
3471                                    input,
3472                                    target,
3473                                );
3474                            }
3475                        }
3476                    }
3477
3478                    pc += 1;
3479                }
3480                OpCode::UpdatePdaReverseLookup {
3481                    state_id: _,
3482                    lookup_name,
3483                    pda_address,
3484                    primary_key,
3485                } => {
3486                    let actual_state_id = override_state_id;
3487                    let pda_val = self.registers[*pda_address].clone();
3488                    let pk_val = self.registers[*primary_key].clone();
3489
3490                    if let (Some(pda_str), Some(pk_str)) = (pda_val.as_str(), pk_val.as_str()) {
3491                        let pending = self.update_pda_reverse_lookup(
3492                            actual_state_id,
3493                            lookup_name,
3494                            pda_str.to_string(),
3495                            pk_str.to_string(),
3496                        )?;
3497                        self.pending_pda_reprocess_updates.extend(pending);
3498                        self.last_pda_registered = Some(pda_str.to_string());
3499                        self.emit_debug(|| VmDebugEvent::PdaReverseLookupUpdate {
3500                            entity_name: entity_name.to_string(),
3501                            event_type: event_type.to_string(),
3502                            lookup_name: lookup_name.clone(),
3503                            pda_address: pda_str.to_string(),
3504                            primary_key: pk_val.clone(),
3505                        });
3506                    } else if !pk_val.is_null() {
3507                        if let Some(pk_num) = pk_val.as_u64() {
3508                            if let Some(pda_str) = pda_val.as_str() {
3509                                let pending = self.update_pda_reverse_lookup(
3510                                    actual_state_id,
3511                                    lookup_name,
3512                                    pda_str.to_string(),
3513                                    pk_num.to_string(),
3514                                )?;
3515                                self.pending_pda_reprocess_updates.extend(pending);
3516                                self.last_pda_registered = Some(pda_str.to_string());
3517                                self.emit_debug(|| VmDebugEvent::PdaReverseLookupUpdate {
3518                                    entity_name: entity_name.to_string(),
3519                                    event_type: event_type.to_string(),
3520                                    lookup_name: lookup_name.clone(),
3521                                    pda_address: pda_str.to_string(),
3522                                    primary_key: pk_val.clone(),
3523                                });
3524                            }
3525                        }
3526                    }
3527
3528                    pc += 1;
3529                }
3530            }
3531
3532            self.instructions_executed += 1;
3533        }
3534
3535        Ok(output)
3536    }
3537
3538    fn load_field(
3539        &self,
3540        event_value: &Value,
3541        path: &FieldPath,
3542        default: Option<&Value>,
3543    ) -> Result<Value> {
3544        if path.segments.is_empty() {
3545            if let Some(obj) = event_value.as_object() {
3546                let filtered: serde_json::Map<String, Value> = obj
3547                    .iter()
3548                    .filter(|(k, _)| !k.starts_with("__"))
3549                    .map(|(k, v)| (k.clone(), v.clone()))
3550                    .collect();
3551                return Ok(Value::Object(filtered));
3552            }
3553            return Ok(event_value.clone());
3554        }
3555
3556        let mut current = event_value;
3557        for segment in path.segments.iter() {
3558            current = match current.get(segment) {
3559                Some(v) => v,
3560                None => {
3561                    tracing::trace!(
3562                        "load_field: segment={:?} not found in {:?}, returning default",
3563                        segment,
3564                        current
3565                    );
3566                    return Ok(default.cloned().unwrap_or(Value::Null));
3567                }
3568            };
3569        }
3570
3571        tracing::trace!("load_field: path={:?}, result={:?}", path.segments, current);
3572        Ok(current.clone())
3573    }
3574
3575    fn get_value_at_path(value: &Value, path: &str) -> Option<Value> {
3576        let mut current = value;
3577        for segment in path.split('.') {
3578            current = current.get(segment)?;
3579        }
3580        Some(current.clone())
3581    }
3582
3583    fn set_field_auto_vivify(
3584        &mut self,
3585        object_reg: Register,
3586        path: &str,
3587        value_reg: Register,
3588    ) -> Result<()> {
3589        let compiled = self.get_compiled_path(path);
3590        let segments = compiled.segments();
3591        let value = self.registers[value_reg].clone();
3592
3593        if !self.registers[object_reg].is_object() {
3594            self.registers[object_reg] = json!({});
3595        }
3596
3597        let obj = self.registers[object_reg]
3598            .as_object_mut()
3599            .ok_or("Not an object")?;
3600
3601        let mut current = obj;
3602        for (i, segment) in segments.iter().enumerate() {
3603            if i == segments.len() - 1 {
3604                current.insert(segment.to_string(), value);
3605                return Ok(());
3606            } else {
3607                current
3608                    .entry(segment.to_string())
3609                    .or_insert_with(|| json!({}));
3610                current = current
3611                    .get_mut(segment)
3612                    .and_then(|v| v.as_object_mut())
3613                    .ok_or("Path collision: expected object")?;
3614            }
3615        }
3616
3617        Ok(())
3618    }
3619
3620    fn set_field_if_null(
3621        &mut self,
3622        object_reg: Register,
3623        path: &str,
3624        value_reg: Register,
3625    ) -> Result<bool> {
3626        let compiled = self.get_compiled_path(path);
3627        let segments = compiled.segments();
3628        let value = self.registers[value_reg].clone();
3629
3630        // SetOnce should only set meaningful values. A null source typically means
3631        // the field doesn't exist in this event type (e.g., instruction events don't
3632        // have account data). Skip to preserve any existing value.
3633        if value.is_null() {
3634            return Ok(false);
3635        }
3636
3637        if !self.registers[object_reg].is_object() {
3638            self.registers[object_reg] = json!({});
3639        }
3640
3641        let obj = self.registers[object_reg]
3642            .as_object_mut()
3643            .ok_or("Not an object")?;
3644
3645        let mut current = obj;
3646        for (i, segment) in segments.iter().enumerate() {
3647            if i == segments.len() - 1 {
3648                if !current.contains_key(segment) || current.get(segment).unwrap().is_null() {
3649                    current.insert(segment.to_string(), value);
3650                    return Ok(true);
3651                }
3652                return Ok(false);
3653            } else {
3654                current
3655                    .entry(segment.to_string())
3656                    .or_insert_with(|| json!({}));
3657                current = current
3658                    .get_mut(segment)
3659                    .and_then(|v| v.as_object_mut())
3660                    .ok_or("Path collision: expected object")?;
3661            }
3662        }
3663
3664        Ok(false)
3665    }
3666
3667    fn set_field_max(
3668        &mut self,
3669        object_reg: Register,
3670        path: &str,
3671        value_reg: Register,
3672    ) -> Result<bool> {
3673        let compiled = self.get_compiled_path(path);
3674        let segments = compiled.segments();
3675        let new_value = self.registers[value_reg].clone();
3676
3677        if !self.registers[object_reg].is_object() {
3678            self.registers[object_reg] = json!({});
3679        }
3680
3681        let obj = self.registers[object_reg]
3682            .as_object_mut()
3683            .ok_or("Not an object")?;
3684
3685        let mut current = obj;
3686        for (i, segment) in segments.iter().enumerate() {
3687            if i == segments.len() - 1 {
3688                let should_update = if let Some(current_value) = current.get(segment) {
3689                    if current_value.is_null() {
3690                        true
3691                    } else {
3692                        match (current_value.as_i64(), new_value.as_i64()) {
3693                            (Some(current_val), Some(new_val)) => new_val > current_val,
3694                            (Some(current_val), None) if new_value.as_u64().is_some() => {
3695                                new_value.as_u64().unwrap() as i64 > current_val
3696                            }
3697                            (None, Some(new_val)) if current_value.as_u64().is_some() => {
3698                                new_val > current_value.as_u64().unwrap() as i64
3699                            }
3700                            (None, None) => match (current_value.as_u64(), new_value.as_u64()) {
3701                                (Some(current_val), Some(new_val)) => new_val > current_val,
3702                                _ => match (current_value.as_f64(), new_value.as_f64()) {
3703                                    (Some(current_val), Some(new_val)) => new_val > current_val,
3704                                    _ => false,
3705                                },
3706                            },
3707                            _ => false,
3708                        }
3709                    }
3710                } else {
3711                    true
3712                };
3713
3714                if should_update {
3715                    current.insert(segment.to_string(), new_value);
3716                    return Ok(true);
3717                }
3718                return Ok(false);
3719            } else {
3720                current
3721                    .entry(segment.to_string())
3722                    .or_insert_with(|| json!({}));
3723                current = current
3724                    .get_mut(segment)
3725                    .and_then(|v| v.as_object_mut())
3726                    .ok_or("Path collision: expected object")?;
3727            }
3728        }
3729
3730        Ok(false)
3731    }
3732
3733    fn set_field_sum(
3734        &mut self,
3735        object_reg: Register,
3736        path: &str,
3737        value_reg: Register,
3738    ) -> Result<bool> {
3739        let compiled = self.get_compiled_path(path);
3740        let segments = compiled.segments();
3741        let new_value = &self.registers[value_reg];
3742
3743        // Extract numeric value before borrowing object_reg mutably
3744        tracing::trace!(
3745            "set_field_sum: path={:?}, value={:?}, value_type={}",
3746            path,
3747            new_value,
3748            match new_value {
3749                serde_json::Value::Null => "null",
3750                serde_json::Value::Bool(_) => "bool",
3751                serde_json::Value::Number(_) => "number",
3752                serde_json::Value::String(_) => "string",
3753                serde_json::Value::Array(_) => "array",
3754                serde_json::Value::Object(_) => "object",
3755            }
3756        );
3757        let new_val_num = new_value
3758            .as_i64()
3759            .or_else(|| new_value.as_u64().map(|n| n as i64))
3760            .ok_or("Sum requires numeric value")?;
3761
3762        if !self.registers[object_reg].is_object() {
3763            self.registers[object_reg] = json!({});
3764        }
3765
3766        let obj = self.registers[object_reg]
3767            .as_object_mut()
3768            .ok_or("Not an object")?;
3769
3770        let mut current = obj;
3771        for (i, segment) in segments.iter().enumerate() {
3772            if i == segments.len() - 1 {
3773                let current_val = current
3774                    .get(segment)
3775                    .and_then(|v| {
3776                        if v.is_null() {
3777                            None
3778                        } else {
3779                            v.as_i64().or_else(|| v.as_u64().map(|n| n as i64))
3780                        }
3781                    })
3782                    .unwrap_or(0);
3783
3784                let sum = current_val + new_val_num;
3785                current.insert(segment.to_string(), json!(sum));
3786                return Ok(true);
3787            } else {
3788                current
3789                    .entry(segment.to_string())
3790                    .or_insert_with(|| json!({}));
3791                current = current
3792                    .get_mut(segment)
3793                    .and_then(|v| v.as_object_mut())
3794                    .ok_or("Path collision: expected object")?;
3795            }
3796        }
3797
3798        Ok(false)
3799    }
3800
3801    fn set_field_increment(&mut self, object_reg: Register, path: &str) -> Result<bool> {
3802        let compiled = self.get_compiled_path(path);
3803        let segments = compiled.segments();
3804
3805        if !self.registers[object_reg].is_object() {
3806            self.registers[object_reg] = json!({});
3807        }
3808
3809        let obj = self.registers[object_reg]
3810            .as_object_mut()
3811            .ok_or("Not an object")?;
3812
3813        let mut current = obj;
3814        for (i, segment) in segments.iter().enumerate() {
3815            if i == segments.len() - 1 {
3816                // Get current value (default to 0 if null/missing)
3817                let current_val = current
3818                    .get(segment)
3819                    .and_then(|v| {
3820                        if v.is_null() {
3821                            None
3822                        } else {
3823                            v.as_i64().or_else(|| v.as_u64().map(|n| n as i64))
3824                        }
3825                    })
3826                    .unwrap_or(0);
3827
3828                let incremented = current_val + 1;
3829                current.insert(segment.to_string(), json!(incremented));
3830                return Ok(true);
3831            } else {
3832                current
3833                    .entry(segment.to_string())
3834                    .or_insert_with(|| json!({}));
3835                current = current
3836                    .get_mut(segment)
3837                    .and_then(|v| v.as_object_mut())
3838                    .ok_or("Path collision: expected object")?;
3839            }
3840        }
3841
3842        Ok(false)
3843    }
3844
3845    fn set_field_min(
3846        &mut self,
3847        object_reg: Register,
3848        path: &str,
3849        value_reg: Register,
3850    ) -> Result<bool> {
3851        let compiled = self.get_compiled_path(path);
3852        let segments = compiled.segments();
3853        let new_value = self.registers[value_reg].clone();
3854
3855        if !self.registers[object_reg].is_object() {
3856            self.registers[object_reg] = json!({});
3857        }
3858
3859        let obj = self.registers[object_reg]
3860            .as_object_mut()
3861            .ok_or("Not an object")?;
3862
3863        let mut current = obj;
3864        for (i, segment) in segments.iter().enumerate() {
3865            if i == segments.len() - 1 {
3866                let should_update = if let Some(current_value) = current.get(segment) {
3867                    if current_value.is_null() {
3868                        true
3869                    } else {
3870                        match (current_value.as_i64(), new_value.as_i64()) {
3871                            (Some(current_val), Some(new_val)) => new_val < current_val,
3872                            (Some(current_val), None) if new_value.as_u64().is_some() => {
3873                                (new_value.as_u64().unwrap() as i64) < current_val
3874                            }
3875                            (None, Some(new_val)) if current_value.as_u64().is_some() => {
3876                                new_val < current_value.as_u64().unwrap() as i64
3877                            }
3878                            (None, None) => match (current_value.as_u64(), new_value.as_u64()) {
3879                                (Some(current_val), Some(new_val)) => new_val < current_val,
3880                                _ => match (current_value.as_f64(), new_value.as_f64()) {
3881                                    (Some(current_val), Some(new_val)) => new_val < current_val,
3882                                    _ => false,
3883                                },
3884                            },
3885                            _ => false,
3886                        }
3887                    }
3888                } else {
3889                    true
3890                };
3891
3892                if should_update {
3893                    current.insert(segment.to_string(), new_value);
3894                    return Ok(true);
3895                }
3896                return Ok(false);
3897            } else {
3898                current
3899                    .entry(segment.to_string())
3900                    .or_insert_with(|| json!({}));
3901                current = current
3902                    .get_mut(segment)
3903                    .and_then(|v| v.as_object_mut())
3904                    .ok_or("Path collision: expected object")?;
3905            }
3906        }
3907
3908        Ok(false)
3909    }
3910
3911    fn get_field(&mut self, object_reg: Register, path: &str) -> Result<Value> {
3912        let compiled = self.get_compiled_path(path);
3913        let segments = compiled.segments();
3914        let mut current = &self.registers[object_reg];
3915
3916        for segment in segments {
3917            current = current
3918                .get(segment)
3919                .ok_or_else(|| format!("Field not found: {}", segment))?;
3920        }
3921
3922        Ok(current.clone())
3923    }
3924
3925    fn append_to_array(
3926        &mut self,
3927        object_reg: Register,
3928        path: &str,
3929        value_reg: Register,
3930        max_length: usize,
3931    ) -> Result<()> {
3932        let compiled = self.get_compiled_path(path);
3933        let segments = compiled.segments();
3934        let value = self.registers[value_reg].clone();
3935
3936        if !self.registers[object_reg].is_object() {
3937            self.registers[object_reg] = json!({});
3938        }
3939
3940        let obj = self.registers[object_reg]
3941            .as_object_mut()
3942            .ok_or("Not an object")?;
3943
3944        let mut current = obj;
3945        for (i, segment) in segments.iter().enumerate() {
3946            if i == segments.len() - 1 {
3947                current
3948                    .entry(segment.to_string())
3949                    .or_insert_with(|| json!([]));
3950                let arr = current
3951                    .get_mut(segment)
3952                    .and_then(|v| v.as_array_mut())
3953                    .ok_or("Path is not an array")?;
3954                arr.push(value.clone());
3955
3956                if arr.len() > max_length {
3957                    let excess = arr.len() - max_length;
3958                    arr.drain(0..excess);
3959                }
3960            } else {
3961                current
3962                    .entry(segment.to_string())
3963                    .or_insert_with(|| json!({}));
3964                current = current
3965                    .get_mut(segment)
3966                    .and_then(|v| v.as_object_mut())
3967                    .ok_or("Path collision: expected object")?;
3968            }
3969        }
3970
3971        Ok(())
3972    }
3973
3974    fn transform_in_place(&mut self, reg: Register, transformation: &Transformation) -> Result<()> {
3975        let value = &self.registers[reg];
3976        let transformed = Self::apply_transformation(value, transformation)?;
3977        self.registers[reg] = transformed;
3978        Ok(())
3979    }
3980
3981    fn apply_transformation(value: &Value, transformation: &Transformation) -> Result<Value> {
3982        match transformation {
3983            Transformation::HexEncode => {
3984                if let Some(arr) = value.as_array() {
3985                    let bytes: Vec<u8> = arr
3986                        .iter()
3987                        .filter_map(|v| v.as_u64().map(|n| n as u8))
3988                        .collect();
3989                    let hex = hex::encode(&bytes);
3990                    Ok(json!(hex))
3991                } else if value.is_string() {
3992                    Ok(value.clone())
3993                } else {
3994                    Err("HexEncode requires an array of numbers".into())
3995                }
3996            }
3997            Transformation::HexDecode => {
3998                if let Some(s) = value.as_str() {
3999                    let s = s.strip_prefix("0x").unwrap_or(s);
4000                    let bytes = hex::decode(s).map_err(|e| format!("Hex decode error: {}", e))?;
4001                    Ok(json!(bytes))
4002                } else {
4003                    Err("HexDecode requires a string".into())
4004                }
4005            }
4006            Transformation::Base58Encode => {
4007                if let Some(arr) = value.as_array() {
4008                    let bytes: Vec<u8> = arr
4009                        .iter()
4010                        .filter_map(|v| v.as_u64().map(|n| n as u8))
4011                        .collect();
4012                    let encoded = bs58::encode(&bytes).into_string();
4013                    Ok(json!(encoded))
4014                } else if value.is_string() {
4015                    Ok(value.clone())
4016                } else {
4017                    Err("Base58Encode requires an array of numbers".into())
4018                }
4019            }
4020            Transformation::Base58Decode => {
4021                if let Some(s) = value.as_str() {
4022                    let bytes = bs58::decode(s)
4023                        .into_vec()
4024                        .map_err(|e| format!("Base58 decode error: {}", e))?;
4025                    Ok(json!(bytes))
4026                } else {
4027                    Err("Base58Decode requires a string".into())
4028                }
4029            }
4030            Transformation::ToString => Ok(json!(value.to_string())),
4031            Transformation::ToNumber => {
4032                if let Some(s) = value.as_str() {
4033                    let n = s
4034                        .parse::<i64>()
4035                        .map_err(|e| format!("Parse error: {}", e))?;
4036                    Ok(json!(n))
4037                } else {
4038                    Ok(value.clone())
4039                }
4040            }
4041        }
4042    }
4043
4044    fn evaluate_comparison(
4045        &self,
4046        field_value: &Value,
4047        op: &ComparisonOp,
4048        condition_value: &Value,
4049    ) -> Result<bool> {
4050        use ComparisonOp::*;
4051
4052        match op {
4053            Equal => Ok(field_value == condition_value),
4054            NotEqual => Ok(field_value != condition_value),
4055            GreaterThan => {
4056                // Try to compare as numbers
4057                match (field_value.as_i64(), condition_value.as_i64()) {
4058                    (Some(a), Some(b)) => Ok(a > b),
4059                    _ => match (field_value.as_u64(), condition_value.as_u64()) {
4060                        (Some(a), Some(b)) => Ok(a > b),
4061                        _ => match (field_value.as_f64(), condition_value.as_f64()) {
4062                            (Some(a), Some(b)) => Ok(a > b),
4063                            _ => Err("Cannot compare non-numeric values with GreaterThan".into()),
4064                        },
4065                    },
4066                }
4067            }
4068            GreaterThanOrEqual => match (field_value.as_i64(), condition_value.as_i64()) {
4069                (Some(a), Some(b)) => Ok(a >= b),
4070                _ => match (field_value.as_u64(), condition_value.as_u64()) {
4071                    (Some(a), Some(b)) => Ok(a >= b),
4072                    _ => match (field_value.as_f64(), condition_value.as_f64()) {
4073                        (Some(a), Some(b)) => Ok(a >= b),
4074                        _ => {
4075                            Err("Cannot compare non-numeric values with GreaterThanOrEqual".into())
4076                        }
4077                    },
4078                },
4079            },
4080            LessThan => match (field_value.as_i64(), condition_value.as_i64()) {
4081                (Some(a), Some(b)) => Ok(a < b),
4082                _ => match (field_value.as_u64(), condition_value.as_u64()) {
4083                    (Some(a), Some(b)) => Ok(a < b),
4084                    _ => match (field_value.as_f64(), condition_value.as_f64()) {
4085                        (Some(a), Some(b)) => Ok(a < b),
4086                        _ => Err("Cannot compare non-numeric values with LessThan".into()),
4087                    },
4088                },
4089            },
4090            LessThanOrEqual => match (field_value.as_i64(), condition_value.as_i64()) {
4091                (Some(a), Some(b)) => Ok(a <= b),
4092                _ => match (field_value.as_u64(), condition_value.as_u64()) {
4093                    (Some(a), Some(b)) => Ok(a <= b),
4094                    _ => match (field_value.as_f64(), condition_value.as_f64()) {
4095                        (Some(a), Some(b)) => Ok(a <= b),
4096                        _ => Err("Cannot compare non-numeric values with LessThanOrEqual".into()),
4097                    },
4098                },
4099            },
4100        }
4101    }
4102
4103    fn can_resolve_further(&self, value: &Value, state_id: u32, index_name: &str) -> bool {
4104        if let Some(state) = self.states.get(&state_id) {
4105            if let Some(index) = state.lookup_indexes.get(index_name) {
4106                if index.lookup(value).is_some() {
4107                    return true;
4108                }
4109            }
4110
4111            for (name, index) in state.lookup_indexes.iter() {
4112                if name == index_name {
4113                    continue;
4114                }
4115                if index.lookup(value).is_some() {
4116                    return true;
4117                }
4118            }
4119
4120            if let Some(pda_str) = value.as_str() {
4121                if let Some(pda_lookup) = state.pda_reverse_lookups.get("default_pda_lookup") {
4122                    if pda_lookup.contains(pda_str) {
4123                        return true;
4124                    }
4125                }
4126            }
4127        }
4128
4129        false
4130    }
4131
4132    #[allow(clippy::type_complexity)]
4133    fn apply_deferred_when_op(
4134        &mut self,
4135        state_id: u32,
4136        op: &DeferredWhenOperation,
4137        entity_evaluator: Option<
4138            &Box<dyn Fn(&mut Value, Option<u64>, i64) -> ComputedEvaluatorResult + Send + Sync>,
4139        >,
4140        computed_paths: Option<&[String]>,
4141    ) -> Result<Vec<Mutation>> {
4142        let state = self.states.get(&state_id).ok_or("State not found")?;
4143
4144        if op.primary_key.is_null() {
4145            return Ok(vec![]);
4146        }
4147
4148        let mut entity_state = state
4149            .get_and_touch(&op.primary_key)
4150            .unwrap_or_else(|| json!({}));
4151
4152        // Track old values of computed fields before setting the new value
4153        let old_computed_values: Vec<_> = computed_paths
4154            .map(|paths| {
4155                paths
4156                    .iter()
4157                    .map(|path| Self::get_value_at_path(&entity_state, path))
4158                    .collect()
4159            })
4160            .unwrap_or_default();
4161
4162        Self::set_nested_field_value(&mut entity_state, &op.field_path, op.field_value.clone())?;
4163
4164        // Re-evaluate computed fields if an evaluator is provided
4165        if let Some(evaluator) = entity_evaluator {
4166            let context_slot = self.current_context.as_ref().and_then(|c| c.slot);
4167            let context_timestamp = self
4168                .current_context
4169                .as_ref()
4170                .map(|c| c.timestamp())
4171                .unwrap_or_else(|| {
4172                    std::time::SystemTime::now()
4173                        .duration_since(std::time::UNIX_EPOCH)
4174                        .unwrap()
4175                        .as_secs() as i64
4176                });
4177
4178            tracing::debug!(
4179                entity_name = %op.entity_name,
4180                primary_key = %op.primary_key,
4181                field_path = %op.field_path,
4182                "Re-evaluating computed fields after deferred when-op"
4183            );
4184
4185            if let Err(e) = evaluator(&mut entity_state, context_slot, context_timestamp) {
4186                tracing::warn!(
4187                    entity_name = %op.entity_name,
4188                    primary_key = %op.primary_key,
4189                    error = %e,
4190                    "Failed to evaluate computed fields after deferred when-op"
4191                );
4192            }
4193        }
4194
4195        state.insert_with_eviction(op.primary_key.clone(), entity_state.clone());
4196
4197        if !op.emit {
4198            return Ok(vec![]);
4199        }
4200
4201        let mut patch = json!({});
4202        Self::set_nested_field_value(&mut patch, &op.field_path, op.field_value.clone())?;
4203
4204        // Add computed field changes to the patch
4205        if let Some(paths) = computed_paths {
4206            tracing::debug!(
4207                entity_name = %op.entity_name,
4208                primary_key = %op.primary_key,
4209                computed_paths_count = paths.len(),
4210                "Checking computed fields for changes after deferred when-op"
4211            );
4212            for (path, old_value) in paths.iter().zip(old_computed_values.iter()) {
4213                let new_value = Self::get_value_at_path(&entity_state, path);
4214                tracing::debug!(
4215                    entity_name = %op.entity_name,
4216                    primary_key = %op.primary_key,
4217                    field_path = %path,
4218                    old_value = ?old_value,
4219                    new_value = ?new_value,
4220                    "Comparing computed field values"
4221                );
4222                if let Some(ref new_val) = new_value {
4223                    if Some(new_val) != old_value.as_ref() {
4224                        Self::set_nested_field_value(&mut patch, path, new_val.clone())?;
4225                        tracing::info!(
4226                            entity_name = %op.entity_name,
4227                            primary_key = %op.primary_key,
4228                            field_path = %path,
4229                            "Computed field changed after deferred when-op, including in mutation"
4230                        );
4231                    }
4232                }
4233            }
4234        }
4235
4236        Ok(vec![Mutation {
4237            export: op.entity_name.clone(),
4238            key: op.primary_key.clone(),
4239            patch,
4240            append: vec![],
4241        }])
4242    }
4243
4244    fn set_nested_field_value(obj: &mut Value, path: &str, value: Value) -> Result<()> {
4245        let parts: Vec<&str> = path.split('.').collect();
4246        let mut current = obj;
4247
4248        for (i, part) in parts.iter().enumerate() {
4249            if i == parts.len() - 1 {
4250                if let Some(map) = current.as_object_mut() {
4251                    map.insert(part.to_string(), value);
4252                    return Ok(());
4253                }
4254                return Err("Cannot set field on non-object".into());
4255            }
4256
4257            if current.get(*part).is_none() || !current.get(*part).unwrap().is_object() {
4258                if let Some(map) = current.as_object_mut() {
4259                    map.insert(part.to_string(), json!({}));
4260                }
4261            }
4262
4263            current = current.get_mut(*part).ok_or("Path navigation failed")?;
4264        }
4265
4266        Ok(())
4267    }
4268
4269    pub fn cleanup_expired_when_ops(&mut self, state_id: u32, max_age_secs: i64) -> usize {
4270        let now = std::time::SystemTime::now()
4271            .duration_since(std::time::UNIX_EPOCH)
4272            .unwrap()
4273            .as_secs() as i64;
4274
4275        let state = match self.states.get(&state_id) {
4276            Some(s) => s,
4277            None => return 0,
4278        };
4279
4280        let mut removed = 0;
4281        state.deferred_when_ops.retain(|_, ops| {
4282            let before = ops.len();
4283            ops.retain(|op| now - op.deferred_at < max_age_secs);
4284            removed += before - ops.len();
4285            !ops.is_empty()
4286        });
4287
4288        removed
4289    }
4290
4291    /// Update a PDA reverse lookup and return pending updates for reprocessing.
4292    /// Returns any pending account updates that were queued for this PDA.
4293    /// ```ignore
4294    /// let pending = vm.update_pda_reverse_lookup(state_id, lookup_name, pda_addr, seed)?;
4295    /// for update in pending {
4296    ///     vm.process_event(&bytecode, update.account_data, &update.account_type, None, None)?;
4297    /// }
4298    /// ```
4299    #[cfg_attr(feature = "otel", instrument(
4300        name = "vm.update_pda_lookup",
4301        skip(self),
4302        fields(
4303            pda = %pda_address,
4304            seed = %seed_value,
4305        )
4306    ))]
4307    pub fn update_pda_reverse_lookup(
4308        &mut self,
4309        state_id: u32,
4310        lookup_name: &str,
4311        pda_address: String,
4312        seed_value: String,
4313    ) -> Result<Vec<PendingAccountUpdate>> {
4314        let state = self
4315            .states
4316            .get_mut(&state_id)
4317            .ok_or("State table not found")?;
4318
4319        let lookup = state
4320            .pda_reverse_lookups
4321            .entry(lookup_name.to_string())
4322            .or_insert_with(|| PdaReverseLookup::new(DEFAULT_MAX_PDA_REVERSE_LOOKUP_ENTRIES));
4323
4324        // Detect if the PDA mapping is CHANGING (same PDA, different seed).
4325        // This happens at round boundaries when e.g. entropyVar is remapped
4326        // from old_round to new_round by a Reset instruction.
4327        let old_seed = lookup.index.peek(&pda_address).cloned();
4328        let mapping_changed = old_seed
4329            .as_ref()
4330            .map(|old| old != &seed_value)
4331            .unwrap_or(false);
4332
4333        if !mapping_changed && old_seed.is_none() {
4334            tracing::info!(
4335                pda = %pda_address,
4336                seed = %seed_value,
4337                "[PDA] First-time PDA reverse lookup established"
4338            );
4339        } else if !mapping_changed {
4340            tracing::debug!(
4341                pda = %pda_address,
4342                seed = %seed_value,
4343                "[PDA] PDA reverse lookup re-registered (same mapping)"
4344            );
4345        }
4346
4347        let evicted_pda = lookup.insert(pda_address.clone(), seed_value.clone());
4348
4349        if let Some(ref evicted) = evicted_pda {
4350            if let Some((_, evicted_updates)) = state.pending_updates.remove(evicted) {
4351                let count = evicted_updates.len();
4352                self.pending_queue_size = self.pending_queue_size.saturating_sub(count as u64);
4353            }
4354        }
4355
4356        // Flush pending updates from QueueUntil for this PDA
4357        let mut pending = self.flush_pending_updates(state_id, &pda_address)?;
4358
4359        // When the mapping changed, the last account update for this PDA was
4360        // processed with the OLD seed (wrong key).  We need to:
4361        // 1. Remove stale lookup-index entries that map this PDA address to the
4362        //    old primary key — otherwise LookupIndex resolves the stale entry
4363        //    before PDA reverse lookup is even tried.
4364        // 2. Pull the cached account data and return it for reprocessing with
4365        //    the new mapping.
4366        if mapping_changed {
4367            if let Some(state) = self.states.get(&state_id) {
4368                // Clear stale lookup-index entries for this PDA address
4369                for index in state.lookup_indexes.values() {
4370                    index.remove(&Value::String(pda_address.clone()));
4371                }
4372
4373                if let Some((_, mut cached)) = state.last_account_data.remove(&pda_address) {
4374                    tracing::info!(
4375                        pda = %pda_address,
4376                        old_seed = ?old_seed,
4377                        new_seed = %seed_value,
4378                        account_type = %cached.account_type,
4379                        "PDA mapping changed — clearing stale indexes and reprocessing cached data"
4380                    );
4381                    cached.is_stale_reprocess = true;
4382                    pending.push(cached);
4383                }
4384            }
4385        }
4386
4387        Ok(pending)
4388    }
4389
4390    /// Clean up expired pending updates that are older than the TTL
4391    ///
4392    /// Returns the number of updates that were removed.
4393    /// This should be called periodically to prevent memory leaks from orphaned updates.
4394    pub fn cleanup_expired_pending_updates(&mut self, state_id: u32) -> usize {
4395        let state = match self.states.get_mut(&state_id) {
4396            Some(s) => s,
4397            None => return 0,
4398        };
4399
4400        let now = std::time::SystemTime::now()
4401            .duration_since(std::time::UNIX_EPOCH)
4402            .unwrap()
4403            .as_secs() as i64;
4404
4405        let mut removed_count = 0;
4406
4407        // Iterate through all pending updates and remove expired ones
4408        state.pending_updates.retain(|_pda_address, updates| {
4409            let original_len = updates.len();
4410
4411            updates.retain(|update| {
4412                let age = now - update.queued_at;
4413                age <= PENDING_UPDATE_TTL_SECONDS
4414            });
4415
4416            removed_count += original_len - updates.len();
4417
4418            // Remove the entry entirely if no updates remain
4419            !updates.is_empty()
4420        });
4421
4422        // Update the global counter
4423        self.pending_queue_size = self.pending_queue_size.saturating_sub(removed_count as u64);
4424
4425        if removed_count > 0 {
4426            #[cfg(feature = "otel")]
4427            crate::vm_metrics::record_pending_updates_expired(
4428                removed_count as u64,
4429                &state.entity_name,
4430            );
4431        }
4432
4433        removed_count
4434    }
4435
4436    /// Queue an account update for later processing when PDA reverse lookup is not yet available
4437    ///
4438    /// # Workflow
4439    ///
4440    /// This implements a deferred processing pattern for account updates when the PDA reverse
4441    /// lookup needed to resolve the primary key is not yet available:
4442    ///
4443    /// 1. **Initial Account Update**: When an account update arrives but the PDA reverse lookup
4444    ///    is not available, call `queue_account_update()` to queue it for later.
4445    ///
4446    /// 2. **Register PDA Mapping**: When the instruction that establishes the PDA mapping is
4447    ///    processed, call `update_pda_reverse_lookup()` which returns pending updates.
4448    ///
4449    /// 3. **Reprocess Pending Updates**: Process the returned pending updates through the VM:
4450    ///    ```ignore
4451    ///    let pending = vm.update_pda_reverse_lookup(state_id, lookup_name, pda_addr, seed)?;
4452    ///    for update in pending {
4453    ///        let mutations = vm.process_event(
4454    ///            &bytecode, update.account_data, &update.account_type, None, None
4455    ///        )?;
4456    ///    }
4457    ///    ```
4458    ///
4459    /// # Arguments
4460    ///
4461    /// * `state_id` - The state table ID
4462    /// * `pda_address` - The PDA address that needs reverse lookup
4463    /// * `account_type` - The event type name for reprocessing
4464    /// * `account_data` - The account data (event value) for reprocessing
4465    /// * `slot` - The slot number when this update occurred
4466    /// * `signature` - The transaction signature
4467    #[cfg_attr(feature = "otel", instrument(
4468        name = "vm.queue_account_update",
4469        skip(self, update),
4470        fields(
4471            pda = %update.pda_address,
4472            account_type = %update.account_type,
4473            slot = update.slot,
4474        )
4475    ))]
4476    pub fn queue_account_update(
4477        &mut self,
4478        state_id: u32,
4479        update: QueuedAccountUpdate,
4480    ) -> Result<()> {
4481        if self.pending_queue_size >= MAX_PENDING_UPDATES_TOTAL as u64 {
4482            self.cleanup_expired_pending_updates(state_id);
4483            if self.pending_queue_size >= MAX_PENDING_UPDATES_TOTAL as u64 {
4484                self.drop_oldest_pending_update(state_id)?;
4485            }
4486        }
4487
4488        let state = self
4489            .states
4490            .get_mut(&state_id)
4491            .ok_or("State table not found")?;
4492
4493        let pending = PendingAccountUpdate {
4494            account_type: update.account_type,
4495            pda_address: update.pda_address.clone(),
4496            account_data: update.account_data,
4497            slot: update.slot,
4498            write_version: update.write_version,
4499            signature: update.signature,
4500            queued_at: std::time::SystemTime::now()
4501                .duration_since(std::time::UNIX_EPOCH)
4502                .unwrap()
4503                .as_secs() as i64,
4504            is_stale_reprocess: false,
4505        };
4506
4507        let pda_address = pending.pda_address.clone();
4508        let slot = pending.slot;
4509
4510        let mut updates = state
4511            .pending_updates
4512            .entry(pda_address.clone())
4513            .or_insert_with(Vec::new);
4514
4515        let original_len = updates.len();
4516        updates.retain(|existing| existing.slot > slot);
4517        let removed_by_dedup = original_len - updates.len();
4518
4519        if removed_by_dedup > 0 {
4520            self.pending_queue_size = self
4521                .pending_queue_size
4522                .saturating_sub(removed_by_dedup as u64);
4523        }
4524
4525        if updates.len() >= MAX_PENDING_UPDATES_PER_PDA {
4526            updates.remove(0);
4527            self.pending_queue_size = self.pending_queue_size.saturating_sub(1);
4528        }
4529
4530        updates.push(pending);
4531        #[cfg(feature = "otel")]
4532        crate::vm_metrics::record_pending_update_queued(&state.entity_name);
4533
4534        Ok(())
4535    }
4536
4537    pub fn queue_instruction_event(
4538        &mut self,
4539        state_id: u32,
4540        event: QueuedInstructionEvent,
4541    ) -> Result<()> {
4542        let state = self
4543            .states
4544            .get_mut(&state_id)
4545            .ok_or("State table not found")?;
4546
4547        let pda_address = event.pda_address.clone();
4548
4549        let pending = PendingInstructionEvent {
4550            event_type: event.event_type,
4551            pda_address: event.pda_address,
4552            event_data: event.event_data,
4553            slot: event.slot,
4554            signature: event.signature,
4555            queued_at: std::time::SystemTime::now()
4556                .duration_since(std::time::UNIX_EPOCH)
4557                .unwrap()
4558                .as_secs() as i64,
4559        };
4560
4561        let mut events = state
4562            .pending_instruction_events
4563            .entry(pda_address)
4564            .or_insert_with(Vec::new);
4565
4566        if events.len() >= MAX_PENDING_UPDATES_PER_PDA {
4567            events.remove(0);
4568        }
4569
4570        events.push(pending);
4571
4572        Ok(())
4573    }
4574
4575    pub fn take_last_pda_lookup_miss(&mut self) -> Option<String> {
4576        self.last_pda_lookup_miss.take()
4577    }
4578
4579    pub fn take_last_lookup_index_miss(&mut self) -> Option<String> {
4580        self.last_lookup_index_miss.take()
4581    }
4582
4583    pub fn take_last_pda_registered(&mut self) -> Option<String> {
4584        self.last_pda_registered.take()
4585    }
4586
4587    pub fn take_last_lookup_index_keys(&mut self) -> Vec<String> {
4588        std::mem::take(&mut self.last_lookup_index_keys)
4589    }
4590
4591    pub fn flush_pending_instruction_events(
4592        &mut self,
4593        state_id: u32,
4594        pda_address: &str,
4595    ) -> Vec<PendingInstructionEvent> {
4596        let state = match self.states.get_mut(&state_id) {
4597            Some(s) => s,
4598            None => return Vec::new(),
4599        };
4600
4601        if let Some((_, events)) = state.pending_instruction_events.remove(pda_address) {
4602            events
4603        } else {
4604            Vec::new()
4605        }
4606    }
4607
4608    /// Get statistics about the pending queue for monitoring
4609    pub fn get_pending_queue_stats(&self, state_id: u32) -> Option<PendingQueueStats> {
4610        let state = self.states.get(&state_id)?;
4611
4612        let now = std::time::SystemTime::now()
4613            .duration_since(std::time::UNIX_EPOCH)
4614            .unwrap()
4615            .as_secs() as i64;
4616
4617        let mut total_updates = 0;
4618        let mut oldest_timestamp = now;
4619        let mut largest_pda_queue = 0;
4620        let mut estimated_memory = 0;
4621
4622        for entry in state.pending_updates.iter() {
4623            let (_, updates) = entry.pair();
4624            total_updates += updates.len();
4625            largest_pda_queue = largest_pda_queue.max(updates.len());
4626
4627            for update in updates.iter() {
4628                oldest_timestamp = oldest_timestamp.min(update.queued_at);
4629                // Rough memory estimate
4630                estimated_memory += update.account_type.len() +
4631                                   update.pda_address.len() +
4632                                   update.signature.len() +
4633                                   16 + // slot + queued_at
4634                                   estimate_json_size(&update.account_data);
4635            }
4636        }
4637
4638        Some(PendingQueueStats {
4639            total_updates,
4640            unique_pdas: state.pending_updates.len(),
4641            oldest_age_seconds: now - oldest_timestamp,
4642            largest_pda_queue_size: largest_pda_queue,
4643            estimated_memory_bytes: estimated_memory,
4644        })
4645    }
4646
4647    pub fn get_memory_stats(&self, state_id: u32) -> VmMemoryStats {
4648        let mut stats = VmMemoryStats {
4649            path_cache_size: self.path_cache.len(),
4650            ..Default::default()
4651        };
4652
4653        if let Some(state) = self.states.get(&state_id) {
4654            stats.state_table_entity_count = state.data.len();
4655            stats.state_table_max_entries = state.config.max_entries;
4656            stats.state_table_at_capacity = state.is_at_capacity();
4657
4658            stats.lookup_index_count = state.lookup_indexes.len();
4659            stats.lookup_index_total_entries =
4660                state.lookup_indexes.values().map(|idx| idx.len()).sum();
4661
4662            stats.temporal_index_count = state.temporal_indexes.len();
4663            stats.temporal_index_total_entries = state
4664                .temporal_indexes
4665                .values()
4666                .map(|idx| idx.total_entries())
4667                .sum();
4668
4669            stats.pda_reverse_lookup_count = state.pda_reverse_lookups.len();
4670            stats.pda_reverse_lookup_total_entries = state
4671                .pda_reverse_lookups
4672                .values()
4673                .map(|lookup| lookup.len())
4674                .sum();
4675
4676            stats.version_tracker_entries = state.version_tracker.len();
4677
4678            stats.pending_queue_stats = self.get_pending_queue_stats(state_id);
4679        }
4680
4681        stats
4682    }
4683
4684    pub fn cleanup_all_expired(&mut self, state_id: u32) -> CleanupResult {
4685        let pending_removed = self.cleanup_expired_pending_updates(state_id);
4686        let temporal_removed = self.cleanup_temporal_indexes(state_id);
4687
4688        #[cfg(feature = "otel")]
4689        if let Some(state) = self.states.get(&state_id) {
4690            crate::vm_metrics::record_cleanup(
4691                pending_removed,
4692                temporal_removed,
4693                &state.entity_name,
4694            );
4695        }
4696
4697        CleanupResult {
4698            pending_updates_removed: pending_removed,
4699            temporal_entries_removed: temporal_removed,
4700        }
4701    }
4702
4703    fn cleanup_temporal_indexes(&mut self, state_id: u32) -> usize {
4704        let state = match self.states.get_mut(&state_id) {
4705            Some(s) => s,
4706            None => return 0,
4707        };
4708
4709        let now = std::time::SystemTime::now()
4710            .duration_since(std::time::UNIX_EPOCH)
4711            .unwrap()
4712            .as_secs() as i64;
4713
4714        let cutoff = now - TEMPORAL_HISTORY_TTL_SECONDS;
4715        let mut total_removed = 0;
4716
4717        for index in state.temporal_indexes.values_mut() {
4718            total_removed += index.cleanup_expired(cutoff);
4719        }
4720
4721        total_removed
4722    }
4723
4724    pub fn check_state_table_capacity(&self, state_id: u32) -> Option<CapacityWarning> {
4725        let state = self.states.get(&state_id)?;
4726
4727        if state.is_at_capacity() {
4728            Some(CapacityWarning {
4729                current_entries: state.data.len(),
4730                max_entries: state.config.max_entries,
4731                entries_over_limit: state.entries_over_limit(),
4732            })
4733        } else {
4734            None
4735        }
4736    }
4737
4738    /// Drop the oldest pending update across all PDAs
4739    fn drop_oldest_pending_update(&mut self, state_id: u32) -> Result<()> {
4740        let state = self
4741            .states
4742            .get_mut(&state_id)
4743            .ok_or("State table not found")?;
4744
4745        let mut oldest_pda: Option<String> = None;
4746        let mut oldest_timestamp = i64::MAX;
4747
4748        // Find the PDA with the oldest update
4749        for entry in state.pending_updates.iter() {
4750            let (pda, updates) = entry.pair();
4751            if let Some(update) = updates.first() {
4752                if update.queued_at < oldest_timestamp {
4753                    oldest_timestamp = update.queued_at;
4754                    oldest_pda = Some(pda.clone());
4755                }
4756            }
4757        }
4758
4759        // Remove the oldest update
4760        if let Some(pda) = oldest_pda {
4761            if let Some(mut updates) = state.pending_updates.get_mut(&pda) {
4762                if !updates.is_empty() {
4763                    updates.remove(0);
4764                    self.pending_queue_size = self.pending_queue_size.saturating_sub(1);
4765
4766                    // Remove the entry if it's now empty
4767                    if updates.is_empty() {
4768                        drop(updates);
4769                        state.pending_updates.remove(&pda);
4770                    }
4771                }
4772            }
4773        }
4774
4775        Ok(())
4776    }
4777
4778    /// Flush and return pending updates for a PDA for external reprocessing
4779    ///
4780    /// Returns the pending updates that were queued for this PDA address.
4781    /// The caller should reprocess these through the VM using process_event().
4782    fn flush_pending_updates(
4783        &mut self,
4784        state_id: u32,
4785        pda_address: &str,
4786    ) -> Result<Vec<PendingAccountUpdate>> {
4787        let state = self
4788            .states
4789            .get_mut(&state_id)
4790            .ok_or("State table not found")?;
4791
4792        if let Some((_, pending_updates)) = state.pending_updates.remove(pda_address) {
4793            let count = pending_updates.len();
4794            self.pending_queue_size = self.pending_queue_size.saturating_sub(count as u64);
4795            #[cfg(feature = "otel")]
4796            crate::vm_metrics::record_pending_updates_flushed(count as u64, &state.entity_name);
4797            Ok(pending_updates)
4798        } else {
4799            Ok(Vec::new())
4800        }
4801    }
4802
4803    /// Cache the most recent account data for a PDA address.
4804    /// Called by the vixen runtime after a Lookup-handler account update is
4805    /// successfully processed.  When a PDA mapping later changes (e.g. at a
4806    /// round boundary), the cached data is returned for reprocessing with the
4807    /// new mapping.
4808    pub fn cache_last_account_data(
4809        &mut self,
4810        state_id: u32,
4811        pda_address: &str,
4812        update: PendingAccountUpdate,
4813    ) {
4814        if let Some(state) = self.states.get(&state_id) {
4815            state
4816                .last_account_data
4817                .insert(pda_address.to_string(), update);
4818        }
4819    }
4820
4821    /// Try to resolve a primary key via PDA reverse lookup
4822    pub fn try_pda_reverse_lookup(
4823        &mut self,
4824        state_id: u32,
4825        lookup_name: &str,
4826        pda_address: &str,
4827    ) -> Option<String> {
4828        let state = self.states.get_mut(&state_id)?;
4829
4830        if let Some(lookup) = state.pda_reverse_lookups.get_mut(lookup_name) {
4831            if let Some(value) = lookup.lookup(pda_address) {
4832                self.pda_cache_hits += 1;
4833                return Some(value);
4834            }
4835        }
4836
4837        self.pda_cache_misses += 1;
4838        None
4839    }
4840
4841    /// Try to resolve a value through lookup indexes.
4842    /// This attempts to resolve the value using any lookup index in the state.
4843    pub fn try_lookup_index_resolution(&self, state_id: u32, value: &Value) -> Option<Value> {
4844        let state = self.states.get(&state_id)?;
4845
4846        for index in state.lookup_indexes.values() {
4847            if let Some(resolved) = index.lookup(value) {
4848                return Some(resolved);
4849            }
4850        }
4851
4852        None
4853    }
4854
4855    /// Try to resolve a primary key via chained PDA + lookup index resolution.
4856    /// First tries PDA reverse lookup, then tries to resolve the result through lookup indexes.
4857    /// This is useful when the PDA maps to an intermediate value (e.g., round address)
4858    /// that needs to be resolved to the actual primary key (e.g., round_id).
4859    pub fn try_chained_pda_lookup(
4860        &mut self,
4861        state_id: u32,
4862        lookup_name: &str,
4863        pda_address: &str,
4864    ) -> Option<String> {
4865        // First, try PDA reverse lookup
4866        let pda_result = self.try_pda_reverse_lookup(state_id, lookup_name, pda_address)?;
4867
4868        // Try to resolve the PDA result through lookup indexes
4869        // (e.g., round address -> round_id)
4870        let pda_value = Value::String(pda_result.clone());
4871        if let Some(resolved) = self.try_lookup_index_resolution(state_id, &pda_value) {
4872            resolved.as_str().map(|s| s.to_string())
4873        } else {
4874            // Return the PDA result if we can't resolve further
4875            pda_value.as_str().map(|s| s.to_string())
4876        }
4877    }
4878
4879    // ============================================================================
4880    // Computed Expression Evaluator (Task 5)
4881    // ============================================================================
4882
4883    /// Evaluate a computed expression AST against the current state
4884    /// This is the core runtime evaluator for computed fields from the AST
4885    pub fn evaluate_computed_expr(&self, expr: &ComputedExpr, state: &Value) -> Result<Value> {
4886        self.evaluate_computed_expr_with_env(expr, state, &std::collections::HashMap::new())
4887    }
4888
4889    /// Evaluate a computed expression with a variable environment (for let bindings)
4890    fn evaluate_computed_expr_with_env(
4891        &self,
4892        expr: &ComputedExpr,
4893        state: &Value,
4894        env: &std::collections::HashMap<String, Value>,
4895    ) -> Result<Value> {
4896        match expr {
4897            ComputedExpr::FieldRef { path } => self.get_field_from_state(state, path),
4898
4899            ComputedExpr::Var { name } => env
4900                .get(name)
4901                .cloned()
4902                .ok_or_else(|| format!("Undefined variable: {}", name).into()),
4903
4904            ComputedExpr::Let { name, value, body } => {
4905                let val = self.evaluate_computed_expr_with_env(value, state, env)?;
4906                let mut new_env = env.clone();
4907                new_env.insert(name.clone(), val);
4908                self.evaluate_computed_expr_with_env(body, state, &new_env)
4909            }
4910
4911            ComputedExpr::If {
4912                condition,
4913                then_branch,
4914                else_branch,
4915            } => {
4916                let cond_val = self.evaluate_computed_expr_with_env(condition, state, env)?;
4917                if self.value_to_bool(&cond_val) {
4918                    self.evaluate_computed_expr_with_env(then_branch, state, env)
4919                } else {
4920                    self.evaluate_computed_expr_with_env(else_branch, state, env)
4921                }
4922            }
4923
4924            ComputedExpr::None => Ok(Value::Null),
4925
4926            ComputedExpr::Some { value } => self.evaluate_computed_expr_with_env(value, state, env),
4927
4928            ComputedExpr::Slice { expr, start, end } => {
4929                let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
4930                match val {
4931                    Value::Array(arr) => {
4932                        let slice: Vec<Value> = arr.get(*start..*end).unwrap_or(&[]).to_vec();
4933                        Ok(Value::Array(slice))
4934                    }
4935                    _ => Err(format!("Cannot slice non-array value: {:?}", val).into()),
4936                }
4937            }
4938
4939            ComputedExpr::Index { expr, index } => {
4940                let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
4941                match val {
4942                    Value::Array(arr) => Ok(arr.get(*index).cloned().unwrap_or(Value::Null)),
4943                    _ => Err(format!("Cannot index non-array value: {:?}", val).into()),
4944                }
4945            }
4946
4947            ComputedExpr::U64FromLeBytes { bytes } => {
4948                let val = self.evaluate_computed_expr_with_env(bytes, state, env)?;
4949                let byte_vec = self.value_to_bytes(&val)?;
4950                if byte_vec.len() < 8 {
4951                    return Err(format!(
4952                        "u64::from_le_bytes requires 8 bytes, got {}",
4953                        byte_vec.len()
4954                    )
4955                    .into());
4956                }
4957                let arr: [u8; 8] = byte_vec[..8]
4958                    .try_into()
4959                    .map_err(|_| "Failed to convert to [u8; 8]")?;
4960                Ok(Value::Number(serde_json::Number::from(u64::from_le_bytes(
4961                    arr,
4962                ))))
4963            }
4964
4965            ComputedExpr::U64FromBeBytes { bytes } => {
4966                let val = self.evaluate_computed_expr_with_env(bytes, state, env)?;
4967                let byte_vec = self.value_to_bytes(&val)?;
4968                if byte_vec.len() < 8 {
4969                    return Err(format!(
4970                        "u64::from_be_bytes requires 8 bytes, got {}",
4971                        byte_vec.len()
4972                    )
4973                    .into());
4974                }
4975                let arr: [u8; 8] = byte_vec[..8]
4976                    .try_into()
4977                    .map_err(|_| "Failed to convert to [u8; 8]")?;
4978                Ok(Value::Number(serde_json::Number::from(u64::from_be_bytes(
4979                    arr,
4980                ))))
4981            }
4982
4983            ComputedExpr::ByteArray { bytes } => {
4984                Ok(Value::Array(bytes.iter().map(|b| json!(*b)).collect()))
4985            }
4986
4987            ComputedExpr::Closure { param, body } => {
4988                // Closures are stored as-is; they're evaluated when used in map()
4989                // Return a special representation
4990                Ok(json!({
4991                    "__closure": {
4992                        "param": param,
4993                        "body": serde_json::to_value(body).unwrap_or(Value::Null)
4994                    }
4995                }))
4996            }
4997
4998            ComputedExpr::Unary { op, expr } => {
4999                let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
5000                self.apply_unary_op(op, &val)
5001            }
5002
5003            ComputedExpr::JsonToBytes { expr } => {
5004                let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
5005                // Convert JSON array of numbers to byte array
5006                let bytes = self.value_to_bytes(&val)?;
5007                Ok(Value::Array(bytes.iter().map(|b| json!(*b)).collect()))
5008            }
5009
5010            ComputedExpr::UnwrapOr { expr, default } => {
5011                let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
5012                if val.is_null() {
5013                    Ok(default.clone())
5014                } else {
5015                    Ok(val)
5016                }
5017            }
5018
5019            ComputedExpr::Binary { op, left, right } => {
5020                let l = self.evaluate_computed_expr_with_env(left, state, env)?;
5021                let r = self.evaluate_computed_expr_with_env(right, state, env)?;
5022                self.apply_binary_op(op, &l, &r)
5023            }
5024
5025            ComputedExpr::Cast { expr, to_type } => {
5026                let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
5027                self.apply_cast(&val, to_type)
5028            }
5029
5030            ComputedExpr::ResolverComputed {
5031                resolver,
5032                method,
5033                args,
5034            } => {
5035                let evaluated_args: Vec<Value> = args
5036                    .iter()
5037                    .map(|arg| self.evaluate_computed_expr_with_env(arg, state, env))
5038                    .collect::<Result<Vec<_>>>()?;
5039                crate::resolvers::evaluate_resolver_computed(resolver, method, &evaluated_args)
5040            }
5041
5042            ComputedExpr::MethodCall { expr, method, args } => {
5043                let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
5044                // Option/iterator methods need the unevaluated closure body.
5045                if (method == "map" || method == "filter") && args.len() == 1 {
5046                    if let ComputedExpr::Closure { param, body } = &args[0] {
5047                        // Option::None.map/filter returns None.
5048                        if val.is_null() {
5049                            return Ok(Value::Null);
5050                        }
5051
5052                        if method == "map" {
5053                            if let Value::Array(arr) = &val {
5054                                let results: Result<Vec<Value>> = arr
5055                                    .iter()
5056                                    .map(|elem| {
5057                                        let mut closure_env = env.clone();
5058                                        closure_env.insert(param.clone(), elem.clone());
5059                                        self.evaluate_computed_expr_with_env(
5060                                            body,
5061                                            state,
5062                                            &closure_env,
5063                                        )
5064                                    })
5065                                    .collect();
5066                                return Ok(Value::Array(results?));
5067                            }
5068
5069                            let mut closure_env = env.clone();
5070                            closure_env.insert(param.clone(), val);
5071                            return self.evaluate_computed_expr_with_env(body, state, &closure_env);
5072                        }
5073
5074                        if let Value::Array(arr) = &val {
5075                            let results: Result<Vec<Value>> = arr
5076                                .iter()
5077                                .filter_map(|elem| {
5078                                    let mut closure_env = env.clone();
5079                                    closure_env.insert(param.clone(), elem.clone());
5080                                    match self.evaluate_computed_expr_with_env(
5081                                        body,
5082                                        state,
5083                                        &closure_env,
5084                                    ) {
5085                                        Ok(predicate) if self.value_to_bool(&predicate) => {
5086                                            Some(Ok(elem.clone()))
5087                                        }
5088                                        Ok(_) => None,
5089                                        Err(error) => Some(Err(error)),
5090                                    }
5091                                })
5092                                .collect();
5093                            return Ok(Value::Array(results?));
5094                        }
5095
5096                        let mut closure_env = env.clone();
5097                        closure_env.insert(param.clone(), val.clone());
5098                        let predicate =
5099                            self.evaluate_computed_expr_with_env(body, state, &closure_env)?;
5100                        return Ok(if self.value_to_bool(&predicate) {
5101                            val
5102                        } else {
5103                            Value::Null
5104                        });
5105                    }
5106                }
5107                let evaluated_args: Vec<Value> = args
5108                    .iter()
5109                    .map(|a| self.evaluate_computed_expr_with_env(a, state, env))
5110                    .collect::<Result<Vec<_>>>()?;
5111                self.apply_method_call(&val, method, &evaluated_args)
5112            }
5113
5114            ComputedExpr::Literal { value } => Ok(value.clone()),
5115
5116            ComputedExpr::Paren { expr } => self.evaluate_computed_expr_with_env(expr, state, env),
5117
5118            ComputedExpr::ContextSlot => Ok(self
5119                .current_context
5120                .as_ref()
5121                .and_then(|ctx| ctx.slot)
5122                .map(|s| json!(s))
5123                .unwrap_or(Value::Null)),
5124
5125            ComputedExpr::ContextTimestamp => Ok(self
5126                .current_context
5127                .as_ref()
5128                .map(|ctx| json!(ctx.timestamp()))
5129                .unwrap_or(Value::Null)),
5130
5131            ComputedExpr::Keccak256 { expr } => {
5132                let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
5133                let bytes = self.value_to_bytes(&val)?;
5134                use sha3::{Digest, Keccak256};
5135                let hash = Keccak256::digest(&bytes);
5136                Ok(Value::Array(
5137                    hash.to_vec().iter().map(|b| json!(*b)).collect(),
5138                ))
5139            }
5140        }
5141    }
5142
5143    /// Convert a JSON value to a byte vector
5144    fn value_to_bytes(&self, val: &Value) -> Result<Vec<u8>> {
5145        match val {
5146            Value::Array(arr) => arr
5147                .iter()
5148                .map(|v| {
5149                    v.as_u64()
5150                        .map(|n| n as u8)
5151                        .ok_or_else(|| "Array element not a valid byte".into())
5152                })
5153                .collect(),
5154            Value::String(s) => {
5155                // Try to decode as hex
5156                if s.starts_with("0x") || s.starts_with("0X") {
5157                    hex::decode(&s[2..]).map_err(|e| format!("Invalid hex string: {}", e).into())
5158                } else {
5159                    hex::decode(s).map_err(|e| format!("Invalid hex string: {}", e).into())
5160                }
5161            }
5162            _ => Err(format!("Cannot convert {:?} to bytes", val).into()),
5163        }
5164    }
5165
5166    /// Apply a unary operation
5167    fn apply_unary_op(&self, op: &crate::ast::UnaryOp, val: &Value) -> Result<Value> {
5168        use crate::ast::UnaryOp;
5169        match op {
5170            UnaryOp::Not => Ok(json!(!self.value_to_bool(val))),
5171            UnaryOp::ReverseBits => match val.as_u64() {
5172                Some(n) => Ok(json!(n.reverse_bits())),
5173                None => match val.as_i64() {
5174                    Some(n) => Ok(json!((n as u64).reverse_bits())),
5175                    None => Err("reverse_bits requires an integer".into()),
5176                },
5177            },
5178        }
5179    }
5180
5181    /// Get a field value from state by path (e.g., "section.field" or just "field")
5182    fn get_field_from_state(&self, state: &Value, path: &str) -> Result<Value> {
5183        let segments: Vec<&str> = path.split('.').collect();
5184        let mut current = state;
5185
5186        for segment in segments {
5187            match current.get(segment) {
5188                Some(v) => current = v,
5189                None => return Ok(Value::Null),
5190            }
5191        }
5192
5193        Ok(current.clone())
5194    }
5195
5196    /// Apply a binary operation to two values
5197    fn apply_binary_op(&self, op: &BinaryOp, left: &Value, right: &Value) -> Result<Value> {
5198        match op {
5199            // Arithmetic operations
5200            BinaryOp::Add => self.numeric_op(left, right, |a, b| a + b, |a, b| a + b),
5201            BinaryOp::Sub => self.numeric_op(left, right, |a, b| a - b, |a, b| a - b),
5202            BinaryOp::Mul => self.numeric_op(left, right, |a, b| a * b, |a, b| a * b),
5203            BinaryOp::Div => {
5204                // Check for division by zero
5205                if let Some(r) = right.as_i64() {
5206                    if r == 0 {
5207                        return Err("Division by zero".into());
5208                    }
5209                }
5210                if let Some(r) = right.as_f64() {
5211                    if r == 0.0 {
5212                        return Err("Division by zero".into());
5213                    }
5214                }
5215                self.numeric_op(left, right, |a, b| a / b, |a, b| a / b)
5216            }
5217            BinaryOp::Mod => {
5218                // Modulo - only for integers
5219                match (left.as_i64(), right.as_i64()) {
5220                    (Some(a), Some(b)) if b != 0 => Ok(json!(a % b)),
5221                    (None, _) | (_, None) => match (left.as_u64(), right.as_u64()) {
5222                        (Some(a), Some(b)) if b != 0 => Ok(json!(a % b)),
5223                        _ => Err("Modulo requires non-zero integer operands".into()),
5224                    },
5225                    _ => Err("Modulo by zero".into()),
5226                }
5227            }
5228
5229            // Comparison operations
5230            BinaryOp::Gt => self.comparison_op(left, right, |a, b| a > b, |a, b| a > b),
5231            BinaryOp::Lt => self.comparison_op(left, right, |a, b| a < b, |a, b| a < b),
5232            BinaryOp::Gte => self.comparison_op(left, right, |a, b| a >= b, |a, b| a >= b),
5233            BinaryOp::Lte => self.comparison_op(left, right, |a, b| a <= b, |a, b| a <= b),
5234            BinaryOp::Eq => Ok(json!(left == right)),
5235            BinaryOp::Ne => Ok(json!(left != right)),
5236
5237            // Logical operations
5238            BinaryOp::And => {
5239                let l_bool = self.value_to_bool(left);
5240                let r_bool = self.value_to_bool(right);
5241                Ok(json!(l_bool && r_bool))
5242            }
5243            BinaryOp::Or => {
5244                let l_bool = self.value_to_bool(left);
5245                let r_bool = self.value_to_bool(right);
5246                Ok(json!(l_bool || r_bool))
5247            }
5248
5249            // Bitwise operations
5250            BinaryOp::Xor => match (left.as_u64(), right.as_u64()) {
5251                (Some(a), Some(b)) => Ok(json!(a ^ b)),
5252                _ => match (left.as_i64(), right.as_i64()) {
5253                    (Some(a), Some(b)) => Ok(json!(a ^ b)),
5254                    _ => Err("XOR requires integer operands".into()),
5255                },
5256            },
5257            BinaryOp::BitAnd => match (left.as_u64(), right.as_u64()) {
5258                (Some(a), Some(b)) => Ok(json!(a & b)),
5259                _ => match (left.as_i64(), right.as_i64()) {
5260                    (Some(a), Some(b)) => Ok(json!(a & b)),
5261                    _ => Err("BitAnd requires integer operands".into()),
5262                },
5263            },
5264            BinaryOp::BitOr => match (left.as_u64(), right.as_u64()) {
5265                (Some(a), Some(b)) => Ok(json!(a | b)),
5266                _ => match (left.as_i64(), right.as_i64()) {
5267                    (Some(a), Some(b)) => Ok(json!(a | b)),
5268                    _ => Err("BitOr requires integer operands".into()),
5269                },
5270            },
5271            BinaryOp::Shl => match (left.as_u64(), right.as_u64()) {
5272                (Some(a), Some(b)) => Ok(json!(a << b)),
5273                _ => match (left.as_i64(), right.as_i64()) {
5274                    (Some(a), Some(b)) => Ok(json!(a << b)),
5275                    _ => Err("Shl requires integer operands".into()),
5276                },
5277            },
5278            BinaryOp::Shr => match (left.as_u64(), right.as_u64()) {
5279                (Some(a), Some(b)) => Ok(json!(a >> b)),
5280                _ => match (left.as_i64(), right.as_i64()) {
5281                    (Some(a), Some(b)) => Ok(json!(a >> b)),
5282                    _ => Err("Shr requires integer operands".into()),
5283                },
5284            },
5285        }
5286    }
5287
5288    /// Helper for numeric operations that can work on integers or floats
5289    fn numeric_op<F1, F2>(
5290        &self,
5291        left: &Value,
5292        right: &Value,
5293        int_op: F1,
5294        float_op: F2,
5295    ) -> Result<Value>
5296    where
5297        F1: Fn(i64, i64) -> i64,
5298        F2: Fn(f64, f64) -> f64,
5299    {
5300        // Try i64 first
5301        if let (Some(a), Some(b)) = (left.as_i64(), right.as_i64()) {
5302            return Ok(json!(int_op(a, b)));
5303        }
5304
5305        // Try u64
5306        if let (Some(a), Some(b)) = (left.as_u64(), right.as_u64()) {
5307            // For u64, we need to be careful with underflow in subtraction
5308            return Ok(json!(int_op(a as i64, b as i64)));
5309        }
5310
5311        // Try f64
5312        if let (Some(a), Some(b)) = (left.as_f64(), right.as_f64()) {
5313            return Ok(json!(float_op(a, b)));
5314        }
5315
5316        // If either is null, return null
5317        if left.is_null() || right.is_null() {
5318            return Ok(Value::Null);
5319        }
5320
5321        Err(format!(
5322            "Cannot perform numeric operation on {:?} and {:?}",
5323            left, right
5324        )
5325        .into())
5326    }
5327
5328    /// Helper for comparison operations
5329    fn comparison_op<F1, F2>(
5330        &self,
5331        left: &Value,
5332        right: &Value,
5333        int_cmp: F1,
5334        float_cmp: F2,
5335    ) -> Result<Value>
5336    where
5337        F1: Fn(i64, i64) -> bool,
5338        F2: Fn(f64, f64) -> bool,
5339    {
5340        // Try i64 first
5341        if let (Some(a), Some(b)) = (left.as_i64(), right.as_i64()) {
5342            return Ok(json!(int_cmp(a, b)));
5343        }
5344
5345        // Try u64
5346        if let (Some(a), Some(b)) = (left.as_u64(), right.as_u64()) {
5347            return Ok(json!(int_cmp(a as i64, b as i64)));
5348        }
5349
5350        // Try f64
5351        if let (Some(a), Some(b)) = (left.as_f64(), right.as_f64()) {
5352            return Ok(json!(float_cmp(a, b)));
5353        }
5354
5355        // If either is null, comparison returns false
5356        if left.is_null() || right.is_null() {
5357            return Ok(json!(false));
5358        }
5359
5360        Err(format!("Cannot compare {:?} and {:?}", left, right).into())
5361    }
5362
5363    /// Convert a value to boolean for logical operations
5364    fn value_to_bool(&self, value: &Value) -> bool {
5365        match value {
5366            Value::Null => false,
5367            Value::Bool(b) => *b,
5368            Value::Number(n) => {
5369                if let Some(i) = n.as_i64() {
5370                    i != 0
5371                } else if let Some(f) = n.as_f64() {
5372                    f != 0.0
5373                } else {
5374                    true
5375                }
5376            }
5377            Value::String(s) => !s.is_empty(),
5378            Value::Array(arr) => !arr.is_empty(),
5379            Value::Object(obj) => !obj.is_empty(),
5380        }
5381    }
5382
5383    /// Apply a type cast to a value
5384    fn apply_cast(&self, value: &Value, to_type: &str) -> Result<Value> {
5385        match to_type {
5386            "i8" | "i16" | "i32" | "i64" | "isize" => {
5387                if let Some(n) = value.as_i64() {
5388                    Ok(json!(n))
5389                } else if let Some(n) = value.as_u64() {
5390                    Ok(json!(n as i64))
5391                } else if let Some(n) = value.as_f64() {
5392                    Ok(json!(n as i64))
5393                } else if let Some(s) = value.as_str() {
5394                    s.parse::<i64>()
5395                        .map(|n| json!(n))
5396                        .map_err(|e| format!("Cannot parse '{}' as integer: {}", s, e).into())
5397                } else {
5398                    Err(format!("Cannot cast {:?} to {}", value, to_type).into())
5399                }
5400            }
5401            "u8" | "u16" | "u32" | "u64" | "usize" => {
5402                if let Some(n) = value.as_u64() {
5403                    Ok(json!(n))
5404                } else if let Some(n) = value.as_i64() {
5405                    Ok(json!(n as u64))
5406                } else if let Some(n) = value.as_f64() {
5407                    Ok(json!(n as u64))
5408                } else if let Some(s) = value.as_str() {
5409                    s.parse::<u64>().map(|n| json!(n)).map_err(|e| {
5410                        format!("Cannot parse '{}' as unsigned integer: {}", s, e).into()
5411                    })
5412                } else {
5413                    Err(format!("Cannot cast {:?} to {}", value, to_type).into())
5414                }
5415            }
5416            "f32" | "f64" => {
5417                if let Some(n) = value.as_f64() {
5418                    Ok(json!(n))
5419                } else if let Some(n) = value.as_i64() {
5420                    Ok(json!(n as f64))
5421                } else if let Some(n) = value.as_u64() {
5422                    Ok(json!(n as f64))
5423                } else if let Some(s) = value.as_str() {
5424                    s.parse::<f64>()
5425                        .map(|n| json!(n))
5426                        .map_err(|e| format!("Cannot parse '{}' as float: {}", s, e).into())
5427                } else {
5428                    Err(format!("Cannot cast {:?} to {}", value, to_type).into())
5429                }
5430            }
5431            "String" | "string" => Ok(json!(value.to_string())),
5432            "bool" => Ok(json!(self.value_to_bool(value))),
5433            _ => {
5434                // Unknown type, return value as-is
5435                Ok(value.clone())
5436            }
5437        }
5438    }
5439
5440    /// Apply a method call to a value
5441    fn apply_method_call(&self, value: &Value, method: &str, args: &[Value]) -> Result<Value> {
5442        match method {
5443            "unwrap_or" => {
5444                if value.is_null() && !args.is_empty() {
5445                    Ok(args[0].clone())
5446                } else {
5447                    Ok(value.clone())
5448                }
5449            }
5450            "unwrap_or_default" => {
5451                if value.is_null() {
5452                    // Return default for common types
5453                    Ok(json!(0))
5454                } else {
5455                    Ok(value.clone())
5456                }
5457            }
5458            "is_some" => Ok(json!(!value.is_null())),
5459            "is_none" => Ok(json!(value.is_null())),
5460            "is_null" => Ok(json!(value.is_null())),
5461            "or" => {
5462                if args.len() != 1 {
5463                    return Err("or() requires exactly one argument".into());
5464                }
5465                if value.is_null() {
5466                    Ok(args[0].clone())
5467                } else {
5468                    Ok(value.clone())
5469                }
5470            }
5471            "abs" => {
5472                if let Some(n) = value.as_i64() {
5473                    Ok(json!(n.abs()))
5474                } else if let Some(n) = value.as_f64() {
5475                    Ok(json!(n.abs()))
5476                } else {
5477                    Err(format!("Cannot call abs() on {:?}", value).into())
5478                }
5479            }
5480            "len" => {
5481                if let Some(s) = value.as_str() {
5482                    Ok(json!(s.len()))
5483                } else if let Some(arr) = value.as_array() {
5484                    Ok(json!(arr.len()))
5485                } else if let Some(obj) = value.as_object() {
5486                    Ok(json!(obj.len()))
5487                } else {
5488                    Err(format!("Cannot call len() on {:?}", value).into())
5489                }
5490            }
5491            "sum" => {
5492                if !args.is_empty() {
5493                    return Err("sum() does not accept arguments".into());
5494                }
5495                if value.is_null() {
5496                    return Ok(Value::Null);
5497                }
5498
5499                let values = value
5500                    .as_array()
5501                    .ok_or_else(|| format!("Cannot call sum() on {:?}", value))?;
5502
5503                let unsigned_values = values
5504                    .iter()
5505                    .map(|item| {
5506                        item.as_u64()
5507                            .or_else(|| item.as_str().and_then(|text| text.parse::<u64>().ok()))
5508                    })
5509                    .collect::<Option<Vec<_>>>();
5510                if let Some(unsigned_values) = unsigned_values {
5511                    let total = unsigned_values
5512                        .iter()
5513                        .try_fold(0_u64, |total, item| total.checked_add(*item));
5514                    return total
5515                        .map(|total| json!(total))
5516                        .ok_or_else(|| "sum() unsigned integer overflow".into());
5517                }
5518
5519                let signed_values = values
5520                    .iter()
5521                    .map(|item| {
5522                        item.as_i64()
5523                            .or_else(|| item.as_str().and_then(|text| text.parse::<i64>().ok()))
5524                    })
5525                    .collect::<Option<Vec<_>>>();
5526                if let Some(signed_values) = signed_values {
5527                    let total = signed_values
5528                        .iter()
5529                        .try_fold(0_i64, |total, item| total.checked_add(*item));
5530                    return total
5531                        .map(|total| json!(total))
5532                        .ok_or_else(|| "sum() signed integer overflow".into());
5533                }
5534
5535                let total = values.iter().try_fold(0.0_f64, |total, item| {
5536                    item.as_f64()
5537                        .or_else(|| item.as_str().and_then(|text| text.parse::<f64>().ok()))
5538                        .map(|value| total + value)
5539                });
5540                match total {
5541                    Some(total) if total.is_finite() => serde_json::Number::from_f64(total)
5542                        .map(Value::Number)
5543                        .ok_or_else(|| "Failed to serialize sum() result".into()),
5544                    Some(_) => Err("sum() result is not finite".into()),
5545                    None => Err(format!("Cannot sum non-numeric array {:?}", value).into()),
5546                }
5547            }
5548            "to_string" => Ok(json!(value.to_string())),
5549            "min" => {
5550                if args.is_empty() {
5551                    return Err("min() requires an argument".into());
5552                }
5553                let other = &args[0];
5554                if let (Some(a), Some(b)) = (value.as_i64(), other.as_i64()) {
5555                    Ok(json!(a.min(b)))
5556                } else if let (Some(a), Some(b)) = (value.as_f64(), other.as_f64()) {
5557                    Ok(json!(a.min(b)))
5558                } else {
5559                    Err(format!("Cannot call min() on {:?} and {:?}", value, other).into())
5560                }
5561            }
5562            "max" => {
5563                if args.is_empty() {
5564                    return Err("max() requires an argument".into());
5565                }
5566                let other = &args[0];
5567                if let (Some(a), Some(b)) = (value.as_i64(), other.as_i64()) {
5568                    Ok(json!(a.max(b)))
5569                } else if let (Some(a), Some(b)) = (value.as_f64(), other.as_f64()) {
5570                    Ok(json!(a.max(b)))
5571                } else {
5572                    Err(format!("Cannot call max() on {:?} and {:?}", value, other).into())
5573                }
5574            }
5575            "saturating_add" => {
5576                if args.is_empty() {
5577                    return Err("saturating_add() requires an argument".into());
5578                }
5579                let other = &args[0];
5580                if let (Some(a), Some(b)) = (value.as_i64(), other.as_i64()) {
5581                    Ok(json!(a.saturating_add(b)))
5582                } else if let (Some(a), Some(b)) = (value.as_u64(), other.as_u64()) {
5583                    Ok(json!(a.saturating_add(b)))
5584                } else {
5585                    Err(format!(
5586                        "Cannot call saturating_add() on {:?} and {:?}",
5587                        value, other
5588                    )
5589                    .into())
5590                }
5591            }
5592            "saturating_sub" => {
5593                if args.is_empty() {
5594                    return Err("saturating_sub() requires an argument".into());
5595                }
5596                let other = &args[0];
5597                if let (Some(a), Some(b)) = (value.as_i64(), other.as_i64()) {
5598                    Ok(json!(a.saturating_sub(b)))
5599                } else if let (Some(a), Some(b)) = (value.as_u64(), other.as_u64()) {
5600                    Ok(json!(a.saturating_sub(b)))
5601                } else {
5602                    Err(format!(
5603                        "Cannot call saturating_sub() on {:?} and {:?}",
5604                        value, other
5605                    )
5606                    .into())
5607                }
5608            }
5609            _ => Err(format!("Unknown method call: {}()", method).into()),
5610        }
5611    }
5612
5613    /// Evaluate all computed fields for an entity and update the state
5614    /// This takes a list of ComputedFieldSpec from the AST and applies them
5615    pub fn evaluate_computed_fields_from_ast(
5616        &self,
5617        state: &mut Value,
5618        computed_field_specs: &[ComputedFieldSpec],
5619    ) -> Result<Vec<String>> {
5620        let mut updated_paths = Vec::new();
5621
5622        crate::resolvers::validate_resolver_computed_specs(computed_field_specs)?;
5623
5624        for spec in computed_field_specs {
5625            match self.evaluate_computed_expr(&spec.expression, state) {
5626                Ok(result) => {
5627                    self.set_field_in_state(state, &spec.target_path, result)?;
5628                    updated_paths.push(spec.target_path.clone());
5629                }
5630                Err(e) => {
5631                    tracing::warn!(
5632                        target_path = %spec.target_path,
5633                        error = %e,
5634                        "Failed to evaluate computed field"
5635                    );
5636                }
5637            }
5638        }
5639
5640        Ok(updated_paths)
5641    }
5642
5643    /// Set a field value in state by path (e.g., "section.field")
5644    fn set_field_in_state(&self, state: &mut Value, path: &str, value: Value) -> Result<()> {
5645        let segments: Vec<&str> = path.split('.').collect();
5646
5647        if segments.is_empty() {
5648            return Err("Empty path".into());
5649        }
5650
5651        // Navigate to parent, creating intermediate objects as needed
5652        let mut current = state;
5653        for (i, segment) in segments.iter().enumerate() {
5654            if i == segments.len() - 1 {
5655                // Last segment - set the value
5656                if let Some(obj) = current.as_object_mut() {
5657                    obj.insert(segment.to_string(), value);
5658                    return Ok(());
5659                } else {
5660                    return Err(format!("Cannot set field '{}' on non-object", segment).into());
5661                }
5662            } else {
5663                // Intermediate segment - navigate or create
5664                if !current.is_object() {
5665                    *current = json!({});
5666                }
5667                let obj = current.as_object_mut().unwrap();
5668                current = obj.entry(segment.to_string()).or_insert_with(|| json!({}));
5669            }
5670        }
5671
5672        Ok(())
5673    }
5674
5675    /// Create a computed fields evaluator closure from AST specs
5676    /// This returns a function that can be passed to the bytecode builder
5677    pub fn create_evaluator_from_specs(
5678        specs: Vec<ComputedFieldSpec>,
5679    ) -> impl Fn(&mut Value, Option<u64>, i64) -> ComputedEvaluatorResult + Send + Sync + 'static
5680    {
5681        move |state: &mut Value, context_slot: Option<u64>, context_timestamp: i64| {
5682            let mut vm = VmContext::new();
5683            vm.current_context = Some(UpdateContext {
5684                slot: context_slot,
5685                timestamp: Some(context_timestamp),
5686                ..Default::default()
5687            });
5688            vm.evaluate_computed_fields_from_ast(state, &specs)
5689                .map_err(|error| {
5690                    Box::<dyn std::error::Error + Send + Sync>::from(std::io::Error::other(
5691                        error.to_string(),
5692                    ))
5693                })?;
5694            Ok(())
5695        }
5696    }
5697}
5698
5699impl Default for VmContext {
5700    fn default() -> Self {
5701        Self::new()
5702    }
5703}
5704
5705// Implement the ReverseLookupUpdater trait for VmContext
5706impl crate::resolvers::ReverseLookupUpdater for VmContext {
5707    fn update(&mut self, pda_address: String, seed_value: String) -> Vec<PendingAccountUpdate> {
5708        // Use default state_id=0 and default lookup name
5709        self.update_pda_reverse_lookup(0, "default_pda_lookup", pda_address, seed_value)
5710            .unwrap_or_else(|e| {
5711                tracing::error!("Failed to update PDA reverse lookup: {}", e);
5712                Vec::new()
5713            })
5714    }
5715
5716    fn flush_pending(&mut self, pda_address: &str) -> Vec<PendingAccountUpdate> {
5717        // Flush is handled inside update_pda_reverse_lookup, but we can also call it directly
5718        self.flush_pending_updates(0, pda_address)
5719            .unwrap_or_else(|e| {
5720                tracing::error!("Failed to flush pending updates: {}", e);
5721                Vec::new()
5722            })
5723    }
5724}
5725
5726#[cfg(test)]
5727mod tests {
5728    use super::*;
5729    use crate::ast::{
5730        BinaryOp, ComputedExpr, ComputedFieldSpec, HttpMethod, UrlResolverConfig, UrlSource,
5731    };
5732
5733    #[test]
5734    fn test_url_resolver_cache_key_uses_method_and_resolved_url() {
5735        let field_path_resolver = ResolverType::Url(UrlResolverConfig {
5736            url_source: UrlSource::FieldPath("metadata_uri".to_string()),
5737            method: HttpMethod::Get,
5738            extract_path: None,
5739        });
5740        let template_resolver = ResolverType::Url(UrlResolverConfig {
5741            url_source: UrlSource::Template(vec![ast::UrlTemplatePart::Literal(
5742                "https://example.com/metadata".to_string(),
5743            )]),
5744            method: HttpMethod::Get,
5745            extract_path: Some("data".to_string()),
5746        });
5747        let input = json!("https://cdn.example.com/token.json");
5748
5749        assert_eq!(
5750            resolver_cache_key(&field_path_resolver, &input),
5751            resolver_cache_key(&template_resolver, &input)
5752        );
5753    }
5754
5755    #[test]
5756    fn test_url_resolver_cache_key_distinguishes_http_method() {
5757        let get_resolver = ResolverType::Url(UrlResolverConfig {
5758            url_source: UrlSource::FieldPath("metadata_uri".to_string()),
5759            method: HttpMethod::Get,
5760            extract_path: None,
5761        });
5762        let post_resolver = ResolverType::Url(UrlResolverConfig {
5763            url_source: UrlSource::FieldPath("metadata_uri".to_string()),
5764            method: HttpMethod::Post,
5765            extract_path: None,
5766        });
5767        let input = json!("https://api.example.com/round");
5768
5769        assert_ne!(
5770            resolver_cache_key(&get_resolver, &input),
5771            resolver_cache_key(&post_resolver, &input)
5772        );
5773    }
5774
5775    #[test]
5776    fn test_expired_resolver_cache_entry_is_dropped() {
5777        let mut vm = VmContext::new();
5778        let resolver = ResolverType::Url(UrlResolverConfig {
5779            url_source: UrlSource::FieldPath("metadata_uri".to_string()),
5780            method: HttpMethod::Get,
5781            extract_path: None,
5782        });
5783        let input = json!("https://cdn.example.com/token.json");
5784        let cache_key = resolver_cache_key(&resolver, &input);
5785
5786        vm.resolver_cache.put(
5787            cache_key.clone(),
5788            ResolverCacheEntry {
5789                value: ResolverCacheValue::Resolved(json!({ "name": "Token" })),
5790                cached_at: Instant::now() - resolver_cache_ttl() - Duration::from_secs(1),
5791                ttl: resolver_cache_ttl(),
5792            },
5793        );
5794
5795        assert!(vm.get_cached_resolver_value(&cache_key).is_none());
5796        assert!(vm.resolver_cache.get(&cache_key).is_none());
5797    }
5798
5799    #[test]
5800    fn test_in_flight_resolver_request_collects_targets_without_requeueing() {
5801        let mut vm = VmContext::new();
5802        let resolver = ResolverType::Token;
5803        let input = json!("mint_123");
5804
5805        vm.enqueue_resolver_request(
5806            "ignored".to_string(),
5807            resolver.clone(),
5808            input.clone(),
5809            ResolverTarget {
5810                state_id: 0,
5811                entity_name: "TestEntity".to_string(),
5812                primary_key: json!("entity_1"),
5813                extracts: vec![],
5814            },
5815        );
5816
5817        let first_batch = vm.take_resolver_requests();
5818        assert_eq!(first_batch.len(), 1);
5819
5820        let cache_key = resolver_cache_key(&resolver, &input);
5821        assert!(vm.resolver_pending.get(&cache_key).unwrap().in_flight);
5822
5823        vm.enqueue_resolver_request(
5824            "ignored-again".to_string(),
5825            resolver,
5826            input,
5827            ResolverTarget {
5828                state_id: 0,
5829                entity_name: "TestEntity".to_string(),
5830                primary_key: json!("entity_2"),
5831                extracts: vec![],
5832            },
5833        );
5834
5835        assert!(vm.take_resolver_requests().is_empty());
5836        let pending = vm.resolver_pending.get(&cache_key).unwrap();
5837        assert!(pending.in_flight);
5838        assert_eq!(pending.targets.len(), 2);
5839    }
5840
5841    #[test]
5842    fn test_computed_field_preserves_integer_type() {
5843        let vm = VmContext::new();
5844
5845        let mut state = serde_json::json!({
5846            "trading": {
5847                "total_buy_volume": 20000000000_i64,
5848                "total_sell_volume": 17951316474_i64
5849            }
5850        });
5851
5852        let spec = ComputedFieldSpec {
5853            target_path: "trading.total_volume".to_string(),
5854            result_type: "Option<u64>".to_string(),
5855            expression: ComputedExpr::Binary {
5856                op: BinaryOp::Add,
5857                left: Box::new(ComputedExpr::UnwrapOr {
5858                    expr: Box::new(ComputedExpr::FieldRef {
5859                        path: "trading.total_buy_volume".to_string(),
5860                    }),
5861                    default: serde_json::json!(0),
5862                }),
5863                right: Box::new(ComputedExpr::UnwrapOr {
5864                    expr: Box::new(ComputedExpr::FieldRef {
5865                        path: "trading.total_sell_volume".to_string(),
5866                    }),
5867                    default: serde_json::json!(0),
5868                }),
5869            },
5870        };
5871
5872        vm.evaluate_computed_fields_from_ast(&mut state, &[spec])
5873            .unwrap();
5874
5875        let total_volume = state
5876            .get("trading")
5877            .and_then(|t| t.get("total_volume"))
5878            .expect("total_volume should exist");
5879
5880        let serialized = serde_json::to_string(total_volume).unwrap();
5881        assert!(
5882            !serialized.contains('.'),
5883            "Integer should not have decimal point: {}",
5884            serialized
5885        );
5886        assert_eq!(
5887            total_volume.as_i64(),
5888            Some(37951316474),
5889            "Value should be correct sum"
5890        );
5891    }
5892
5893    #[test]
5894    fn test_computed_array_sum_preserves_u64() {
5895        let vm = VmContext::new();
5896        let mut state = json!({
5897            "state": {
5898                "deployed": [422874702_u64, 569506895_u64, 573312366_u64]
5899            }
5900        });
5901        let spec = ComputedFieldSpec {
5902            target_path: "state.total_deployed".to_string(),
5903            result_type: "Option<u64>".to_string(),
5904            expression: ComputedExpr::MethodCall {
5905                expr: Box::new(ComputedExpr::FieldRef {
5906                    path: "state.deployed".to_string(),
5907                }),
5908                method: "sum".to_string(),
5909                args: Vec::new(),
5910            },
5911        };
5912
5913        vm.evaluate_computed_fields_from_ast(&mut state, &[spec])
5914            .unwrap();
5915
5916        assert_eq!(state["state"]["total_deployed"], json!(1_565_693_963_u64));
5917    }
5918
5919    #[test]
5920    fn test_computed_array_sum_accepts_u64_strings() {
5921        let vm = VmContext::new();
5922        let mut state = json!({
5923            "state": {
5924                "deployed": ["422874702", "569506895", "573312366"]
5925            }
5926        });
5927        let spec = ComputedFieldSpec {
5928            target_path: "state.total_deployed".to_string(),
5929            result_type: "Option<u64>".to_string(),
5930            expression: ComputedExpr::MethodCall {
5931                expr: Box::new(ComputedExpr::FieldRef {
5932                    path: "state.deployed".to_string(),
5933                }),
5934                method: "sum".to_string(),
5935                args: Vec::new(),
5936            },
5937        };
5938
5939        vm.evaluate_computed_fields_from_ast(&mut state, &[spec])
5940            .unwrap();
5941
5942        assert_eq!(state["state"]["total_deployed"], json!(1_565_693_963_u64));
5943    }
5944
5945    #[test]
5946    fn test_computed_array_sum_propagates_null() {
5947        let vm = VmContext::new();
5948        let mut state = json!({ "state": { "deployed": null } });
5949        let spec = ComputedFieldSpec {
5950            target_path: "state.total_deployed".to_string(),
5951            result_type: "Option<u64>".to_string(),
5952            expression: ComputedExpr::MethodCall {
5953                expr: Box::new(ComputedExpr::FieldRef {
5954                    path: "state.deployed".to_string(),
5955                }),
5956                method: "sum".to_string(),
5957                args: Vec::new(),
5958            },
5959        };
5960
5961        vm.evaluate_computed_fields_from_ast(&mut state, &[spec])
5962            .unwrap();
5963
5964        assert!(state["state"]["total_deployed"].is_null());
5965    }
5966
5967    #[test]
5968    fn test_computed_option_filter_or_fallback() {
5969        fn specs() -> Vec<ComputedFieldSpec> {
5970            vec![
5971                ComputedFieldSpec {
5972                    target_path: "results.pre_reveal_rng".to_string(),
5973                    result_type: "Option<u64>".to_string(),
5974                    expression: ComputedExpr::MethodCall {
5975                        expr: Box::new(ComputedExpr::MethodCall {
5976                            expr: Box::new(ComputedExpr::FieldRef {
5977                                path: "results.rng".to_string(),
5978                            }),
5979                            method: "filter".to_string(),
5980                            args: vec![ComputedExpr::Closure {
5981                                param: "rng".to_string(),
5982                                body: Box::new(ComputedExpr::Unary {
5983                                    op: crate::ast::UnaryOp::Not,
5984                                    expr: Box::new(ComputedExpr::MethodCall {
5985                                        expr: Box::new(ComputedExpr::Var {
5986                                            name: "rng".to_string(),
5987                                        }),
5988                                        method: "is_null".to_string(),
5989                                        args: Vec::new(),
5990                                    }),
5991                                }),
5992                            }],
5993                        }),
5994                        method: "or".to_string(),
5995                        args: vec![ComputedExpr::FieldRef {
5996                            path: "results.pre_reveal_rng_candidate".to_string(),
5997                        }],
5998                    },
5999                },
6000                ComputedFieldSpec {
6001                    target_path: "results.pre_reveal_winning_square".to_string(),
6002                    result_type: "Option<u64>".to_string(),
6003                    expression: ComputedExpr::MethodCall {
6004                        expr: Box::new(ComputedExpr::FieldRef {
6005                            path: "results.pre_reveal_rng".to_string(),
6006                        }),
6007                        method: "map".to_string(),
6008                        args: vec![ComputedExpr::Closure {
6009                            param: "rng".to_string(),
6010                            body: Box::new(ComputedExpr::Binary {
6011                                op: BinaryOp::Mod,
6012                                left: Box::new(ComputedExpr::Var {
6013                                    name: "rng".to_string(),
6014                                }),
6015                                right: Box::new(ComputedExpr::Literal { value: json!(25) }),
6016                            }),
6017                        }],
6018                    },
6019                },
6020            ]
6021        }
6022
6023        let vm = VmContext::new();
6024        for (rng, candidate, expected_rng, expected_square) in [
6025            (Value::Null, json!(42_u64), json!(42_u64), json!(17_u64)),
6026            (json!(7_u64), json!(42_u64), json!(7_u64), json!(7_u64)),
6027        ] {
6028            let mut state = json!({
6029                "results": {
6030                    "rng": rng,
6031                    "pre_reveal_rng_candidate": candidate,
6032                }
6033            });
6034
6035            vm.evaluate_computed_fields_from_ast(&mut state, &specs())
6036                .unwrap();
6037
6038            assert_eq!(state["results"]["pre_reveal_rng"], expected_rng);
6039            assert_eq!(
6040                state["results"]["pre_reveal_winning_square"],
6041                expected_square
6042            );
6043        }
6044    }
6045
6046    #[test]
6047    fn test_set_field_sum_preserves_integer_type() {
6048        let mut vm = VmContext::new();
6049        vm.registers[0] = serde_json::json!({});
6050        vm.registers[1] = serde_json::json!(20000000000_i64);
6051        vm.registers[2] = serde_json::json!(17951316474_i64);
6052
6053        vm.set_field_sum(0, "trading.total_buy_volume", 1).unwrap();
6054        vm.set_field_sum(0, "trading.total_sell_volume", 2).unwrap();
6055
6056        let state = &vm.registers[0];
6057        let buy_vol = state
6058            .get("trading")
6059            .and_then(|t| t.get("total_buy_volume"))
6060            .unwrap();
6061        let sell_vol = state
6062            .get("trading")
6063            .and_then(|t| t.get("total_sell_volume"))
6064            .unwrap();
6065
6066        let buy_serialized = serde_json::to_string(buy_vol).unwrap();
6067        let sell_serialized = serde_json::to_string(sell_vol).unwrap();
6068
6069        assert!(
6070            !buy_serialized.contains('.'),
6071            "Buy volume should not have decimal: {}",
6072            buy_serialized
6073        );
6074        assert!(
6075            !sell_serialized.contains('.'),
6076            "Sell volume should not have decimal: {}",
6077            sell_serialized
6078        );
6079    }
6080
6081    #[test]
6082    fn test_lookup_index_chaining() {
6083        let mut vm = VmContext::new();
6084
6085        let state = vm.states.get_mut(&0).unwrap();
6086
6087        state
6088            .pda_reverse_lookups
6089            .entry("default_pda_lookup".to_string())
6090            .or_insert_with(|| PdaReverseLookup::new(1000))
6091            .insert("pda_123".to_string(), "addr_456".to_string());
6092
6093        state
6094            .lookup_indexes
6095            .entry("round_address_lookup_index".to_string())
6096            .or_insert_with(LookupIndex::new)
6097            .insert(json!("addr_456"), json!(789));
6098
6099        let handler = vec![
6100            OpCode::LoadConstant {
6101                value: json!("pda_123"),
6102                dest: 0,
6103            },
6104            OpCode::LookupIndex {
6105                state_id: 0,
6106                index_name: "round_address_lookup_index".to_string(),
6107                lookup_value: 0,
6108                dest: 1,
6109            },
6110        ];
6111
6112        vm.execute_handler(&handler, &json!({}), "test", 0, "TestEntity", None, None)
6113            .unwrap();
6114
6115        assert_eq!(vm.registers[1], json!(789));
6116    }
6117
6118    #[test]
6119    fn test_lookup_index_no_chain() {
6120        let mut vm = VmContext::new();
6121
6122        let state = vm.states.get_mut(&0).unwrap();
6123        state
6124            .lookup_indexes
6125            .entry("test_index".to_string())
6126            .or_insert_with(LookupIndex::new)
6127            .insert(json!("key_abc"), json!(42));
6128
6129        let handler = vec![
6130            OpCode::LoadConstant {
6131                value: json!("key_abc"),
6132                dest: 0,
6133            },
6134            OpCode::LookupIndex {
6135                state_id: 0,
6136                index_name: "test_index".to_string(),
6137                lookup_value: 0,
6138                dest: 1,
6139            },
6140        ];
6141
6142        vm.execute_handler(&handler, &json!({}), "test", 0, "TestEntity", None, None)
6143            .unwrap();
6144
6145        assert_eq!(vm.registers[1], json!(42));
6146    }
6147
6148    #[test]
6149    fn test_conditional_set_field_with_zero_array() {
6150        let mut vm = VmContext::new();
6151
6152        let event_zeros = json!({
6153            "value": [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
6154                      0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
6155        });
6156
6157        let event_nonzero = json!({
6158            "value": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16,
6159                      17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32]
6160        });
6161
6162        let zero_32: Value = json!([
6163            0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
6164            0, 0, 0
6165        ]);
6166
6167        let handler = vec![
6168            OpCode::CreateObject { dest: 2 },
6169            OpCode::LoadEventField {
6170                path: FieldPath::new(&["value"]),
6171                dest: 10,
6172                default: None,
6173            },
6174            OpCode::ConditionalSetField {
6175                object: 2,
6176                path: "captured_value".to_string(),
6177                value: 10,
6178                condition_field: FieldPath::new(&["value"]),
6179                condition_op: ComparisonOp::NotEqual,
6180                condition_value: zero_32,
6181            },
6182        ];
6183
6184        vm.execute_handler(&handler, &event_zeros, "test", 0, "Test", None, None)
6185            .unwrap();
6186        assert!(
6187            vm.registers[2].get("captured_value").is_none(),
6188            "Field should not be set when value is all zeros"
6189        );
6190
6191        vm.reset_registers();
6192        vm.execute_handler(&handler, &event_nonzero, "test", 0, "Test", None, None)
6193            .unwrap();
6194        assert!(
6195            vm.registers[2].get("captured_value").is_some(),
6196            "Field should be set when value is non-zero"
6197        );
6198    }
6199
6200    #[test]
6201    fn test_when_instruction_arrives_first() {
6202        let mut vm = VmContext::new();
6203
6204        let signature = "test_sig_123".to_string();
6205
6206        {
6207            let state = vm.states.get(&0).unwrap();
6208            let mut cache = state.recent_tx_instructions.lock().unwrap();
6209            let mut set = HashSet::new();
6210            set.insert("RevealIxState".to_string());
6211            cache.put(signature.clone(), set);
6212        }
6213
6214        vm.current_context = Some(UpdateContext::new(100, signature.clone()));
6215
6216        let handler = vec![
6217            OpCode::CreateObject { dest: 2 },
6218            OpCode::LoadConstant {
6219                value: json!("primary_key_value"),
6220                dest: 1,
6221            },
6222            OpCode::LoadConstant {
6223                value: json!("the_revealed_value"),
6224                dest: 10,
6225            },
6226            OpCode::SetFieldWhen {
6227                object: 2,
6228                path: "entropy_value".to_string(),
6229                value: 10,
6230                when_instruction: "RevealIxState".to_string(),
6231                entity_name: "TestEntity".to_string(),
6232                key_reg: 1,
6233                condition_field: None,
6234                condition_op: None,
6235                condition_value: None,
6236            },
6237        ];
6238
6239        vm.execute_handler(
6240            &handler,
6241            &json!({}),
6242            "VarState",
6243            0,
6244            "TestEntity",
6245            None,
6246            None,
6247        )
6248        .unwrap();
6249
6250        assert_eq!(
6251            vm.registers[2].get("entropy_value").unwrap(),
6252            "the_revealed_value",
6253            "Field should be set when instruction was already seen"
6254        );
6255    }
6256
6257    #[test]
6258    fn test_when_account_arrives_first() {
6259        let mut vm = VmContext::new();
6260
6261        let signature = "test_sig_456".to_string();
6262
6263        vm.current_context = Some(UpdateContext::new(100, signature.clone()));
6264
6265        let handler = vec![
6266            OpCode::CreateObject { dest: 2 },
6267            OpCode::LoadConstant {
6268                value: json!("pk_123"),
6269                dest: 1,
6270            },
6271            OpCode::LoadConstant {
6272                value: json!("deferred_value"),
6273                dest: 10,
6274            },
6275            OpCode::SetFieldWhen {
6276                object: 2,
6277                path: "entropy_value".to_string(),
6278                value: 10,
6279                when_instruction: "RevealIxState".to_string(),
6280                entity_name: "TestEntity".to_string(),
6281                key_reg: 1,
6282                condition_field: None,
6283                condition_op: None,
6284                condition_value: None,
6285            },
6286        ];
6287
6288        vm.execute_handler(
6289            &handler,
6290            &json!({}),
6291            "VarState",
6292            0,
6293            "TestEntity",
6294            None,
6295            None,
6296        )
6297        .unwrap();
6298
6299        assert!(
6300            vm.registers[2].get("entropy_value").is_none(),
6301            "Field should not be set when instruction hasn't been seen"
6302        );
6303
6304        let state = vm.states.get(&0).unwrap();
6305        let key = (signature.clone(), "RevealIxState".to_string());
6306        assert!(
6307            state.deferred_when_ops.contains_key(&key),
6308            "Operation should be queued"
6309        );
6310
6311        {
6312            let mut cache = state.recent_tx_instructions.lock().unwrap();
6313            let mut set = HashSet::new();
6314            set.insert("RevealIxState".to_string());
6315            cache.put(signature.clone(), set);
6316        }
6317
6318        let deferred = state.deferred_when_ops.remove(&key).unwrap().1;
6319        for op in deferred {
6320            vm.apply_deferred_when_op(0, &op, None, None).unwrap();
6321        }
6322
6323        let state = vm.states.get(&0).unwrap();
6324        let entity = state.data.get(&json!("pk_123")).unwrap();
6325        assert_eq!(
6326            entity.get("entropy_value").unwrap(),
6327            "deferred_value",
6328            "Field should be set after instruction arrives"
6329        );
6330    }
6331
6332    #[test]
6333    fn test_when_cleanup_expired() {
6334        let mut vm = VmContext::new();
6335
6336        let state = vm.states.get(&0).unwrap();
6337        let key = ("old_sig".to_string(), "SomeIxState".to_string());
6338        state.deferred_when_ops.insert(
6339            key,
6340            vec![DeferredWhenOperation {
6341                entity_name: "Test".to_string(),
6342                primary_key: json!("pk"),
6343                field_path: "field".to_string(),
6344                field_value: json!("value"),
6345                when_instruction: "SomeIxState".to_string(),
6346                signature: "old_sig".to_string(),
6347                slot: 0,
6348                deferred_at: 0,
6349                emit: true,
6350            }],
6351        );
6352
6353        let removed = vm.cleanup_expired_when_ops(0, 60);
6354
6355        assert_eq!(removed, 1, "Should have removed 1 expired op");
6356        assert!(
6357            vm.states.get(&0).unwrap().deferred_when_ops.is_empty(),
6358            "Deferred ops should be empty after cleanup"
6359        );
6360    }
6361
6362    #[test]
6363    fn test_deferred_when_op_recomputes_dependent_fields() {
6364        use crate::ast::{BinaryOp, ComputedExpr, ComputedFieldSpec};
6365
6366        let mut vm = VmContext::new();
6367
6368        // Create computed field specs similar to the ore stack:
6369        // pre_reveal_rng depends on base_value
6370        // pre_reveal_winning_square depends on pre_reveal_rng
6371        let computed_specs = vec![
6372            ComputedFieldSpec {
6373                target_path: "results.pre_reveal_rng".to_string(),
6374                result_type: "Option<u64>".to_string(),
6375                expression: ComputedExpr::FieldRef {
6376                    path: "entropy.base_value".to_string(),
6377                },
6378            },
6379            ComputedFieldSpec {
6380                target_path: "results.pre_reveal_winning_square".to_string(),
6381                result_type: "Option<u64>".to_string(),
6382                expression: ComputedExpr::MethodCall {
6383                    expr: Box::new(ComputedExpr::FieldRef {
6384                        path: "results.pre_reveal_rng".to_string(),
6385                    }),
6386                    method: "map".to_string(),
6387                    args: vec![ComputedExpr::Closure {
6388                        param: "r".to_string(),
6389                        body: Box::new(ComputedExpr::Binary {
6390                            op: BinaryOp::Mod,
6391                            left: Box::new(ComputedExpr::Var {
6392                                name: "r".to_string(),
6393                            }),
6394                            right: Box::new(ComputedExpr::Literal {
6395                                value: serde_json::json!(25),
6396                            }),
6397                        }),
6398                    }],
6399                },
6400            },
6401        ];
6402
6403        let evaluator: Box<
6404            dyn Fn(&mut Value, Option<u64>, i64) -> ComputedEvaluatorResult + Send + Sync,
6405        > = Box::new(VmContext::create_evaluator_from_specs(computed_specs));
6406
6407        // Test that when we set entropy.base_value via deferred when-op,
6408        // both computed fields are updated
6409        let primary_key = json!("test_pk");
6410        let op = DeferredWhenOperation {
6411            entity_name: "TestEntity".to_string(),
6412            primary_key: primary_key.clone(),
6413            field_path: "entropy.base_value".to_string(),
6414            field_value: json!(100),
6415            when_instruction: "TestIxState".to_string(),
6416            signature: "test_sig".to_string(),
6417            slot: 100,
6418            deferred_at: 0,
6419            emit: true,
6420        };
6421
6422        // Store the entity in state first
6423        let initial_state = json!({
6424            "results": {}
6425        });
6426        vm.states
6427            .get(&0)
6428            .unwrap()
6429            .insert_with_eviction(primary_key.clone(), initial_state);
6430
6431        // Apply the deferred when-op with the evaluator
6432        let mutations = vm
6433            .apply_deferred_when_op(
6434                0,
6435                &op,
6436                Some(&evaluator),
6437                Some(&[
6438                    "results.pre_reveal_rng".to_string(),
6439                    "results.pre_reveal_winning_square".to_string(),
6440                ]),
6441            )
6442            .unwrap();
6443
6444        // Check the entity state
6445        let state = vm.states.get(&0).unwrap();
6446        let entity = state.data.get(&primary_key).unwrap();
6447
6448        println!(
6449            "Entity state: {}",
6450            serde_json::to_string_pretty(&*entity).unwrap()
6451        );
6452
6453        // Verify base value was set
6454        assert_eq!(
6455            entity.get("entropy").and_then(|e| e.get("base_value")),
6456            Some(&json!(100)),
6457            "Base value should be set"
6458        );
6459
6460        // Verify computed fields were calculated
6461        let pre_reveal_rng = entity
6462            .get("results")
6463            .and_then(|r| r.get("pre_reveal_rng"))
6464            .cloned();
6465        let pre_reveal_winning_square = entity
6466            .get("results")
6467            .and_then(|r| r.get("pre_reveal_winning_square"))
6468            .cloned();
6469
6470        assert_eq!(
6471            pre_reveal_rng,
6472            Some(json!(100)),
6473            "pre_reveal_rng should be computed"
6474        );
6475        assert_eq!(
6476            pre_reveal_winning_square,
6477            Some(json!(0)),
6478            "pre_reveal_winning_square should be 100 % 25 = 0"
6479        );
6480
6481        // Verify mutations include computed fields
6482        assert!(!mutations.is_empty(), "Should have mutations");
6483        let mutation = &mutations[0];
6484        let patch = &mutation.patch;
6485
6486        assert!(
6487            patch
6488                .get("results")
6489                .and_then(|r| r.get("pre_reveal_rng"))
6490                .is_some(),
6491            "Mutation should include pre_reveal_rng"
6492        );
6493        assert!(
6494            patch
6495                .get("results")
6496                .and_then(|r| r.get("pre_reveal_winning_square"))
6497                .is_some(),
6498            "Mutation should include pre_reveal_winning_square"
6499        );
6500    }
6501}