arete_interpreter/

compiler.rs

use crate::ast::*;
use serde::Serialize;
use serde_json::Value;
use std::collections::{HashMap, HashSet};
use tracing;

pub type Register = usize;

/// Register holding a write guard's result (see [`OpCode::SkipUnless`]). Kept
/// clear of the registers mappings, keys and captures use.
const GUARD_REG: Register = 30;

fn stop_field_path(target_path: &str) -> String {
    format!("__stop:{}", target_path)
}

/// One independently keyed handler program for an event type.
struct HandlerSegment {
    /// Canonical encoding of the key-loading opcodes of the handlers merged
    /// into this segment. `None` for segments created from an instruction hook
    /// alone.
    key_identity: Option<Value>,
    /// The event field this segment routes by, used to attach instruction
    /// hooks to the segment that shares their `lookup_by`.
    route_field: Option<FieldPath>,
    ops: Vec<OpCode>,
}

/// The event field a key resolution strategy routes by, if any.
fn route_field(resolution: &KeyResolutionStrategy) -> Option<FieldPath> {
    let field = match resolution {
        KeyResolutionStrategy::Embedded { primary_field }
        | KeyResolutionStrategy::Lookup { primary_field, .. }
        | KeyResolutionStrategy::Computed { primary_field, .. } => primary_field,
        KeyResolutionStrategy::TemporalLookup { lookup_field, .. } => lookup_field,
    };
    (!field.segments.is_empty()).then(|| field.clone())
}

/// Merge the mapping opcodes of `new` into `existing`, which resolve the same
/// key: keep the setup (key loading and state read) of `existing`, append the
/// mappings of both, and keep one teardown (state write and mutation).
fn merge_handler_opcodes(existing: &mut Vec<OpCode>, new: &[OpCode]) {
    // Split existing handler into: setup, mappings, teardown
    let mut existing_setup = Vec::new();
    let mut existing_mappings = Vec::new();
    let mut existing_teardown = Vec::new();
    let mut section = 0; // 0=setup, 1=mappings, 2=teardown

    for opcode in existing.iter() {
        match opcode {
            OpCode::ReadOrInitState { .. } => {
                existing_setup.push(opcode.clone());
                section = 1; // Next opcodes are mappings
            }
            OpCode::UpdateState { .. } => {
                existing_teardown.push(opcode.clone());
                section = 2; // Next opcodes are teardown
            }
            OpCode::EmitMutation { .. } => {
                existing_teardown.push(opcode.clone());
            }
            _ if section == 0 => existing_setup.push(opcode.clone()),
            _ if section == 1 => existing_mappings.push(opcode.clone()),
            _ => existing_teardown.push(opcode.clone()),
        }
    }

    // Extract mappings from new handler (skip setup and teardown)
    let mut new_mappings = Vec::new();
    section = 0;

    for opcode in new.iter() {
        match opcode {
            OpCode::ReadOrInitState { .. } => {
                section = 1; // Start capturing mappings
            }
            OpCode::UpdateState { .. } | OpCode::EmitMutation { .. } => {
                section = 2; // Stop capturing
            }
            _ if section == 1 => {
                new_mappings.push(opcode.clone());
            }
            _ => {} // Skip setup and teardown from new handler
        }
    }

    // Rebuild: setup + existing_mappings + new_mappings + teardown
    let mut merged = Vec::new();
    merged.extend(existing_setup);
    merged.extend(existing_mappings);
    merged.extend(new_mappings);
    merged.extend(existing_teardown);

    *existing = merged;
}

fn is_false(value: &bool) -> bool {
    !*value
}

/// The write guard over the mappings of one handler to `field`: `None` when
/// any of them is unguarded (or none maps it), else their guards joined with
/// `||`.
fn index_guard<S>(mappings: &[TypedFieldMapping<S>], field: &str) -> Option<ComputedExpr> {
    let mut guards = mappings
        .iter()
        .filter(|mapping| mapping.target_path == field)
        .map(
            |mapping| match mapping.condition.as_ref()?.parsed.as_ref()? {
                ParsedCondition::Guard { expr } => Some(expr.clone()),
                _ => None,
            },
        )
        .collect::<Option<Vec<_>>>()?
        .into_iter();
    let first = guards.next()?;
    Some(guards.fold(first, |left, right| ComputedExpr::Binary {
        op: BinaryOp::Or,
        left: Box::new(left),
        right: Box::new(right),
    }))
}

/// The lookup index a handler declares it resolves through, if any.
fn declared_index(resolution: &KeyResolutionStrategy) -> Option<&str> {
    match resolution {
        KeyResolutionStrategy::Lookup {
            index_name: Some(index_name),
            ..
        } => Some(index_name),
        _ => None,
    }
}

#[derive(Debug, Clone, Serialize)]
pub enum OpCode {
    /// Abort the handler with empty mutations when the key register is null
    /// and the event is an account-state update (not IxState / CpiEvent).
    /// Placed immediately after key resolution so that downstream opcodes
    /// (index updates, field mappings, resolvers, emit) never execute with
    /// a garbage key. The null-key event is then eligible for queueing by
    /// process_event's miss-handling logic.
    AbortIfNullKey {
        key: Register,
        is_account_event: bool,
    },
    /// Evaluate an expression against the event into `dest`; null when the
    /// expression fails. `transform` applies to a non-null result only, so a
    /// null key stays null.
    EvaluateEventExpr {
        expr: ComputedExpr,
        transform: Option<Transformation>,
        dest: Register,
    },
    LoadEventField {
        path: FieldPath,
        dest: Register,
        default: Option<Value>,
    },
    LoadConstant {
        value: Value,
        dest: Register,
    },
    CopyRegister {
        source: Register,
        dest: Register,
    },
    /// Copy from source to dest only if dest is currently null
    CopyRegisterIfNull {
        source: Register,
        dest: Register,
    },
    GetEventType {
        dest: Register,
    },
    CreateObject {
        dest: Register,
    },
    SetField {
        object: Register,
        path: String,
        value: Register,
    },
    SetFields {
        object: Register,
        fields: Vec<(String, Register)>,
    },
    GetField {
        object: Register,
        path: String,
        dest: Register,
    },
    ReadOrInitState {
        state_id: u32,
        key: Register,
        default: Value,
        dest: Register,
        /// The declared route (lookup index) of this handler. A duplicate
        /// instruction is detected per route when set, so a second route of
        /// the same instruction to the same row is not skipped. `None` for
        /// every handler without a declared lookup, which keeps the
        /// row-and-event-type deduplication.
        #[serde(skip_serializing_if = "Option::is_none")]
        dedup_route: Option<String>,
    },
    UpdateState {
        state_id: u32,
        key: Register,
        value: Register,
    },
    AppendToArray {
        object: Register,
        path: String,
        value: Register,
    },
    /// Load the Unix-seconds timestamp of the update being processed (its
    /// `UpdateContext` timestamp, else the wall clock). It is the update's
    /// time, not the processing time, so a replay reproduces it.
    GetCurrentTimestamp {
        dest: Register,
    },
    /// Wrap an event as `{timestamp, data, slot?, signature?, event_index?,
    /// ix_path?}`, stamped with the update's timestamp.
    CreateEvent {
        dest: Register,
        event_value: Register,
    },
    /// Wrap a captured account as `{timestamp, account_address, data, slot?,
    /// signature?}`, stamped with the update's timestamp.
    CreateCapture {
        dest: Register,
        capture_value: Register,
    },
    Transform {
        source: Register,
        dest: Register,
        transformation: Transformation,
    },
    EmitMutation {
        entity_name: String,
        key: Register,
        state: Register,
    },
    SetFieldIfNull {
        object: Register,
        path: String,
        value: Register,
    },
    SetFieldMax {
        object: Register,
        path: String,
        value: Register,
    },
    UpdateTemporalIndex {
        state_id: u32,
        index_name: String,
        lookup_value: Register,
        primary_key: Register,
        timestamp: Register,
    },
    LookupTemporalIndex {
        state_id: u32,
        index_name: String,
        lookup_value: Register,
        timestamp: Register,
        dest: Register,
    },
    UpdateLookupIndex {
        state_id: u32,
        index_name: String,
        lookup_value: Register,
        primary_key: Register,
    },
    LookupIndex {
        state_id: u32,
        index_name: String,
        lookup_value: Register,
        dest: Register,
        /// A declared lookup: consult only `index_name`, directly or after
        /// PDA reverse lookups, never the entity's other lookup indexes.
        #[serde(skip_serializing_if = "is_false")]
        strict: bool,
    },
    /// Sum a numeric value to a field (accumulator)
    SetFieldSum {
        object: Register,
        path: String,
        value: Register,
    },
    /// Increment a counter field by 1
    SetFieldIncrement {
        object: Register,
        path: String,
    },
    /// Set field to minimum value
    SetFieldMin {
        object: Register,
        path: String,
        value: Register,
    },
    /// Set field only if a specific instruction type was seen in the same transaction.
    /// If not seen yet, defers the operation for later completion.
    SetFieldWhen {
        object: Register,
        path: String,
        value: Register,
        when_instruction: String,
        entity_name: String,
        key_reg: Register,
        condition_field: Option<FieldPath>,
        condition_op: Option<ComparisonOp>,
        condition_value: Option<Value>,
    },
    /// Set field unless stopped by a specific instruction.
    /// Stop is tracked by a per-entity stop flag.
    SetFieldUnlessStopped {
        object: Register,
        path: String,
        value: Register,
        stop_field: String,
        stop_instruction: String,
        entity_name: String,
        key_reg: Register,
    },
    /// Add value to unique set and update count
    /// Maintains internal Set, field stores count
    AddToUniqueSet {
        state_id: u32,
        set_name: String,
        value: Register,
        count_object: Register,
        count_path: String,
    },
    /// Conditionally set a field based on a comparison
    ConditionalSetField {
        object: Register,
        path: String,
        value: Register,
        condition_field: FieldPath,
        condition_op: ComparisonOp,
        condition_value: Value,
    },
    /// Conditionally increment a field based on a comparison
    ConditionalIncrement {
        object: Register,
        path: String,
        condition_field: FieldPath,
        condition_op: ComparisonOp,
        condition_value: Value,
    },
    /// Evaluate computed fields (calls external hook if provided)
    /// computed_paths: List of paths that will be computed (for dirty tracking)
    EvaluateComputedFields {
        state: Register,
        computed_paths: Vec<String>,
    },
    /// Queue a resolver for asynchronous enrichment
    QueueResolver {
        state_id: u32,
        entity_name: String,
        resolver: ResolverType,
        input_path: Option<String>,
        input_value: Option<Value>,
        url_template: Option<Vec<UrlTemplatePart>>,
        strategy: ResolveStrategy,
        extracts: Vec<ResolverExtractSpec>,
        condition: Option<ResolverCondition>,
        schedule_at: Option<String>,
        state: Register,
        key: Register,
    },
    /// Update PDA reverse lookup table
    /// Maps a PDA address to its primary key for reverse lookups
    UpdatePdaReverseLookup {
        state_id: u32,
        lookup_name: String,
        pda_address: Register,
        primary_key: Register,
    },
    /// Skip the next `count` opcodes unless `predicate` holds exactly `true`.
    ///
    /// Emitted only for write-gating guards (`ParsedCondition::Guard`): the
    /// guard is evaluated into `predicate` and the guarded mapping's opcodes
    /// follow, so a `false`, `null` or failed guard leaves the row untouched.
    SkipUnless {
        predicate: Register,
        count: usize,
    },
    /// Remember the fields `paths` of this instruction for the transaction's
    /// sibling registrations, then complete the registrations waiting for an
    /// instruction of this type (see [`SiblingSelector`]).
    CaptureSibling {
        state_id: u32,
        paths: Vec<FieldPath>,
    },
    /// Register PDA mappings read partly or wholly from sibling instructions
    /// of `sibling.instruction_type` in the same transaction. The fields read
    /// from this instruction are in `pda_address` / `primary_key`; the ones
    /// read from a sibling are paths. Siblings that have not arrived yet are
    /// registered when they do.
    RegisterPdaFromSibling {
        state_id: u32,
        lookup_name: String,
        sibling: SiblingSelector,
        pda_field: FieldPath,
        seed_field: FieldPath,
        pda_address: Register,
        primary_key: Register,
    },
    /// Separates independently keyed handler segments for one event type.
    ///
    /// When one event updates the same entity through different keys (for
    /// example `split_position` touching both its `first_position` and its
    /// `second_position`), each key gets its own segment: its own key loading,
    /// state read, mappings, state write and mutation. The VM runs segments in
    /// order and isolates them, so an early exit in one segment never skips
    /// another. Handlers with a single key never contain a boundary.
    SegmentBoundary,
}

pub struct EntityBytecode {
    pub state_id: u32,
    pub handlers: HashMap<String, Vec<OpCode>>,
    pub entity_name: String,
    pub when_events: HashSet<String>,
    pub non_emitted_fields: HashSet<String>,
    pub computed_paths: Vec<String>,
    /// Optional callback for evaluating computed fields
    /// Parameters: state, context_slot (Option<u64>), context_timestamp (i64)
    #[allow(clippy::type_complexity)]
    pub computed_fields_evaluator: Option<
        Box<
            dyn Fn(
                    &mut Value,
                    Option<u64>,
                    i64,
                )
                    -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
                + Send
                + Sync,
        >,
    >,
}

impl std::fmt::Debug for EntityBytecode {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EntityBytecode")
            .field("state_id", &self.state_id)
            .field("handlers", &self.handlers)
            .field("entity_name", &self.entity_name)
            .field("when_events", &self.when_events)
            .field("non_emitted_fields", &self.non_emitted_fields)
            .field("computed_paths", &self.computed_paths)
            .field(
                "computed_fields_evaluator",
                &self.computed_fields_evaluator.is_some(),
            )
            .finish()
    }
}

#[derive(Debug)]
pub struct MultiEntityBytecode {
    pub entities: HashMap<String, EntityBytecode>,
    pub event_routing: HashMap<String, Vec<String>>,
    pub when_events: HashSet<String>,
    pub proto_router: crate::proto_router::ProtoRouter,
}

impl MultiEntityBytecode {
    pub fn from_single<S>(entity_name: String, spec: TypedStreamSpec<S>, state_id: u32) -> Self {
        let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
        let entity_bytecode = compiler.compile_entity();

        let mut entities = HashMap::new();
        let mut event_routing = HashMap::new();
        let mut when_events = HashSet::new();

        for event_type in entity_bytecode.handlers.keys() {
            event_routing
                .entry(event_type.clone())
                .or_insert_with(Vec::new)
                .push(entity_name.clone());
        }

        when_events.extend(entity_bytecode.when_events.iter().cloned());

        entities.insert(entity_name, entity_bytecode);

        MultiEntityBytecode {
            entities,
            event_routing,
            when_events,
            proto_router: crate::proto_router::ProtoRouter::new(),
        }
    }

    pub fn from_entities(entities_vec: Vec<(String, Box<dyn std::any::Any>, u32)>) -> Self {
        let entities = HashMap::new();
        let event_routing = HashMap::new();
        let when_events = HashSet::new();

        if let Some((_entity_name, _spec_any, _state_id)) = entities_vec.into_iter().next() {
            panic!("from_entities requires type information - use builder pattern instead");
        }

        MultiEntityBytecode {
            entities,
            event_routing,
            when_events,
            proto_router: crate::proto_router::ProtoRouter::new(),
        }
    }

    #[allow(clippy::new_ret_no_self)]
    pub fn new() -> MultiEntityBytecodeBuilder {
        MultiEntityBytecodeBuilder {
            entities: HashMap::new(),
            event_routing: HashMap::new(),
            when_events: HashSet::new(),
            proto_router: crate::proto_router::ProtoRouter::new(),
        }
    }

    /// Deterministic content hash of the compiled bytecode, used to invalidate
    /// state snapshots when the stack's compiled logic changes.
    ///
    /// Covers state ids, handler opcode streams, routing, and field metadata.
    /// Computed-field evaluators are opaque closures, so only their presence is
    /// hashed — their bodies are generated from the same stack source as the
    /// opcode streams, which change alongside them in practice.
    ///
    /// Opcode streams are hashed via a canonical serde encoding (externally
    /// tagged variants, object keys sorted), NOT via `Debug` formatting, so the
    /// hash survives cosmetic churn that doesn't alter the compiled program:
    /// rustc `Debug` formatting changes, struct field reordering, and inserting
    /// new `OpCode` variants. Renaming a variant or field DOES change the hash
    /// (names are the tags) — treat renames as snapshot-invalidating. Interpreter
    /// behavior changes that keep the encoding identical are invisible here;
    /// those must bump `SNAPSHOT_FORMAT_VERSION` instead.
    pub fn fingerprint(&self) -> String {
        use sha2::{Digest, Sha256};

        fn update_sorted_set(hasher: &mut Sha256, set: &HashSet<String>) {
            let mut items: Vec<&String> = set.iter().collect();
            items.sort();
            for item in items {
                hasher.update(item.as_bytes());
                hasher.update([0u8]);
            }
            hasher.update([1u8]);
        }

        /// Hash a JSON value in a canonical form: object keys visited in sorted
        /// order with explicit framing bytes. Deliberately independent of
        /// `serde_json::Map`'s iteration order, which flips to insertion order
        /// if any crate in the enclosing workspace enables the `preserve_order`
        /// feature — the fingerprint must agree across differently-featured
        /// builds of the same stack.
        fn update_canonical_json(hasher: &mut Sha256, value: &Value) {
            match value {
                Value::Object(map) => {
                    let mut keys: Vec<&String> = map.keys().collect();
                    keys.sort();
                    hasher.update(*b"{");
                    for key in keys {
                        hasher.update(key.as_bytes());
                        hasher.update([0u8]);
                        update_canonical_json(hasher, &map[key]);
                    }
                    hasher.update(*b"}");
                }
                Value::Array(items) => {
                    hasher.update(*b"[");
                    for item in items {
                        update_canonical_json(hasher, item);
                        hasher.update([0u8]);
                    }
                    hasher.update(*b"]");
                }
                // Scalars: serde_json's rendering (itoa/ryu for numbers, JSON
                // string escaping) is deterministic across platforms.
                scalar => {
                    hasher.update(scalar.to_string().as_bytes());
                    hasher.update([0u8]);
                }
            }
        }

        let mut hasher = Sha256::new();

        let mut entity_names: Vec<&String> = self.entities.keys().collect();
        entity_names.sort();
        for entity_name in entity_names {
            let entity = &self.entities[entity_name];
            hasher.update(entity_name.as_bytes());
            hasher.update([0u8]);
            hasher.update(entity.state_id.to_le_bytes());

            let mut event_types: Vec<&String> = entity.handlers.keys().collect();
            event_types.sort();
            for event_type in event_types {
                hasher.update(event_type.as_bytes());
                hasher.update([0u8]);
                let ops = serde_json::to_value(&entity.handlers[event_type]).expect(
                    "opcode stream must serialize for fingerprinting; \
                     a non-serializable OpCode payload breaks snapshot invalidation",
                );
                update_canonical_json(&mut hasher, &ops);
                hasher.update([0u8]);
            }
            hasher.update([1u8]);

            update_sorted_set(&mut hasher, &entity.when_events);
            update_sorted_set(&mut hasher, &entity.non_emitted_fields);
            for path in &entity.computed_paths {
                hasher.update(path.as_bytes());
                hasher.update([0u8]);
            }
            hasher.update([entity.computed_fields_evaluator.is_some() as u8]);
        }

        let mut routes: Vec<(&String, &Vec<String>)> = self.event_routing.iter().collect();
        routes.sort_by_key(|(event_type, _)| *event_type);
        for (event_type, entity_names) in routes {
            hasher.update(event_type.as_bytes());
            hasher.update([0u8]);
            let mut sorted_names = entity_names.clone();
            sorted_names.sort();
            for name in sorted_names {
                hasher.update(name.as_bytes());
                hasher.update([0u8]);
            }
            hasher.update([1u8]);
        }

        update_sorted_set(&mut hasher, &self.when_events);

        hex::encode(hasher.finalize())
    }
}

pub struct MultiEntityBytecodeBuilder {
    entities: HashMap<String, EntityBytecode>,
    event_routing: HashMap<String, Vec<String>>,
    when_events: HashSet<String>,
    proto_router: crate::proto_router::ProtoRouter,
}

impl MultiEntityBytecodeBuilder {
    pub fn add_entity<S>(
        self,
        entity_name: String,
        spec: TypedStreamSpec<S>,
        state_id: u32,
    ) -> Self {
        self.add_entity_with_evaluator(
            entity_name,
            spec,
            state_id,
            None::<
                fn(
                    &mut Value,
                    Option<u64>,
                    i64,
                )
                    -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>,
            >,
        )
    }

    pub fn add_entity_with_evaluator<S, F>(
        mut self,
        entity_name: String,
        spec: TypedStreamSpec<S>,
        state_id: u32,
        evaluator: Option<F>,
    ) -> Self
    where
        F: Fn(
                &mut Value,
                Option<u64>,
                i64,
            ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
            + Send
            + Sync
            + 'static,
    {
        let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
        let mut entity_bytecode = compiler.compile_entity();

        // Store the evaluator callback if provided
        if let Some(eval) = evaluator {
            entity_bytecode.computed_fields_evaluator = Some(Box::new(eval));
        }

        for event_type in entity_bytecode.handlers.keys() {
            self.event_routing
                .entry(event_type.clone())
                .or_default()
                .push(entity_name.clone());
        }

        self.when_events
            .extend(entity_bytecode.when_events.iter().cloned());

        self.entities.insert(entity_name, entity_bytecode);
        self
    }

    pub fn build(self) -> MultiEntityBytecode {
        MultiEntityBytecode {
            entities: self.entities,
            event_routing: self.event_routing,
            when_events: self.when_events,
            proto_router: self.proto_router,
        }
    }
}

pub struct TypedCompiler<S> {
    pub spec: TypedStreamSpec<S>,
    entity_name: String,
    state_id: u32,
}

impl<S> TypedCompiler<S> {
    pub fn new(spec: TypedStreamSpec<S>, entity_name: String) -> Self {
        TypedCompiler {
            spec,
            entity_name,
            state_id: 0,
        }
    }

    pub fn with_state_id(mut self, state_id: u32) -> Self {
        self.state_id = state_id;
        self
    }

    pub fn compile(&self) -> MultiEntityBytecode {
        let entity_bytecode = self.compile_entity();

        let mut entities = HashMap::new();
        let mut event_routing = HashMap::new();
        let mut when_events = HashSet::new();

        for event_type in entity_bytecode.handlers.keys() {
            event_routing
                .entry(event_type.clone())
                .or_insert_with(Vec::new)
                .push(self.entity_name.clone());
        }

        when_events.extend(entity_bytecode.when_events.iter().cloned());

        entities.insert(self.entity_name.clone(), entity_bytecode);

        MultiEntityBytecode {
            entities,
            event_routing,
            when_events,
            proto_router: crate::proto_router::ProtoRouter::new(),
        }
    }

    fn resolver_outputs_primary_key_directly(&self, primary_field: &FieldPath) -> bool {
        if primary_field.segments.as_slice() != ["__account_address"] {
            return false;
        }

        let primary_key_leafs: HashSet<&str> = self
            .spec
            .identity
            .primary_keys
            .iter()
            .map(|path| path.rsplit('.').next().unwrap_or(path.as_str()))
            .collect();

        self.spec.instruction_hooks.iter().any(|hook| {
            hook.actions.iter().any(|action| {
                matches!(action, HookAction::RegisterPdaMapping { seed_field, .. }
                    if seed_field
                        .segments
                        .last()
                        .is_some_and(|segment| primary_key_leafs.contains(segment.as_str())))
            })
        })
    }

    fn compile_entity(&self) -> EntityBytecode {
        let mut when_events: HashSet<String> = HashSet::new();
        let mut emit_by_path: HashMap<String, bool> = HashMap::new();

        // DEBUG: Collect all handler info before processing
        let mut debug_info = Vec::new();
        for (index, handler_spec) in self.spec.handlers.iter().enumerate() {
            let event_type = self.get_event_type(&handler_spec.source);
            let program_id = match &handler_spec.source {
                crate::ast::SourceSpec::Source { program_id, .. } => {
                    program_id.as_ref().map(|s| s.as_str()).unwrap_or("null")
                }
            };
            debug_info.push(format!(
                "  [{}] EventType={}, Mappings={}, ProgramId={}",
                index,
                event_type,
                handler_spec.mappings.len(),
                program_id
            ));
        }

        // DEBUG: Log handler information (optional - can be removed later)
        // Uncomment to debug handler processing:
        // if self.entity_name == "PumpfunToken" {
        //     eprintln!("🔍 Compiling {} handlers for {}", self.spec.handlers.len(), self.entity_name);
        //     for info in &debug_info {
        //         eprintln!("{}", info);
        //     }
        // }

        // Handlers for one event type are grouped into independently keyed
        // segments. Two handlers merge only when their key resolution compiles
        // to identical opcodes; a handler that resolves its key differently
        // (e.g. `split_position` via `first_position` vs `second_position`)
        // gets its own segment so each update lands on its own entity.
        let mut segments: HashMap<String, Vec<HandlerSegment>> = HashMap::new();

        for handler_spec in &self.spec.handlers {
            for mapping in &handler_spec.mappings {
                if let Some(when) = &mapping.when {
                    when_events.insert(when.clone());
                }
                let entry = emit_by_path
                    .entry(mapping.target_path.clone())
                    .or_insert(false);
                *entry |= mapping.emit;
                if mapping.stop.is_some() {
                    emit_by_path
                        .entry(stop_field_path(&mapping.target_path))
                        .or_insert(false);
                }
            }
            let opcodes = self.compile_handler(handler_spec);
            let event_type = self.get_event_type(&handler_spec.source);
            let key_identity = self.key_identity(handler_spec);

            let event_segments = segments.entry(event_type).or_default();
            if let Some(existing) = event_segments
                .iter_mut()
                .find(|segment| segment.key_identity.as_ref() == Some(&key_identity))
            {
                merge_handler_opcodes(&mut existing.ops, &opcodes);
            } else {
                event_segments.push(HandlerSegment {
                    key_identity: Some(key_identity),
                    route_field: route_field(&handler_spec.key_resolution),
                    ops: opcodes,
                });
            }
        }

        // Process instruction_hooks to add SetField/IncrementField operations
        for hook in &self.spec.instruction_hooks {
            let event_type = hook.instruction_type.clone();
            let event_segments = segments.entry(event_type).or_default();

            // A hook joins the segment that routes by its own `lookup_by` field.
            // Hooks without `lookup_by` (PDA registrations, conditional
            // aggregates) join the first segment. A hook whose `lookup_by`
            // matches no segment gets its own segment keyed by that field
            // instead of silently riding another segment's key.
            let segment_index = match &hook.lookup_by {
                None if !event_segments.is_empty() => Some(0),
                None => None,
                Some(lookup_by) => event_segments
                    .iter()
                    .position(|segment| segment.route_field.as_ref() == Some(lookup_by)),
            };
            let segment_index = match segment_index {
                Some(index) => index,
                None => {
                    // Next to handlers that route differently, resolve the
                    // hook's own key the way a handler would: directly for a
                    // primary-key field, through the lookup index otherwise.
                    // A hook alone keeps the historical direct-key segment.
                    let resolution = if event_segments.is_empty() {
                        None
                    } else {
                        hook.lookup_by
                            .as_ref()
                            .and_then(|lookup_by| self.hook_key_resolution(lookup_by))
                    };
                    let ops = match resolution {
                        Some(resolution) => self.compile_keyed_hook_segment(&resolution),
                        None => self.compile_hook_segment(hook),
                    };
                    event_segments.push(HandlerSegment {
                        key_identity: None,
                        route_field: hook.lookup_by.clone(),
                        ops,
                    });
                    event_segments.len() - 1
                }
            };
            let handler_opcodes = &mut event_segments[segment_index].ops;

            // Generate opcodes for each action in the hook
            let hook_opcodes = self.compile_instruction_hook_actions(&hook.actions);

            // Insert hook opcodes before EvaluateComputedFields (if present) or UpdateState
            // Hook actions (like whale_trade_count increment) must run before computed fields
            // are evaluated, since computed fields may depend on the modified state
            let insert_pos = handler_opcodes
                .iter()
                .position(|op| matches!(op, OpCode::EvaluateComputedFields { .. }))
                .or_else(|| {
                    handler_opcodes
                        .iter()
                        .position(|op| matches!(op, OpCode::UpdateState { .. }))
                });

            if let Some(pos) = insert_pos {
                // Insert hook opcodes before EvaluateComputedFields or UpdateState
                for (i, opcode) in hook_opcodes.into_iter().enumerate() {
                    handler_opcodes.insert(pos + i, opcode);
                }
            }
        }

        // Sibling registrations read fields of other instructions in the
        // transaction: each such instruction captures them first thing, in a
        // segment of its own, before any of its handlers resolve keys through
        // the registrations it completes.
        for (event_type, paths) in self.sibling_capture_paths() {
            let event_segments = segments.entry(event_type).or_default();
            event_segments.insert(
                0,
                HandlerSegment {
                    key_identity: None,
                    route_field: None,
                    ops: vec![OpCode::CaptureSibling {
                        state_id: self.state_id,
                        paths,
                    }],
                },
            );
        }

        let handlers: HashMap<String, Vec<OpCode>> = segments
            .into_iter()
            .map(|(event_type, event_segments)| {
                let mut ops = Vec::new();
                for (index, segment) in event_segments.into_iter().enumerate() {
                    if index > 0 {
                        ops.push(OpCode::SegmentBoundary);
                    }
                    ops.extend(segment.ops);
                }
                (event_type, ops)
            })
            .collect();

        let non_emitted_fields: HashSet<String> = emit_by_path
            .into_iter()
            .filter_map(|(path, emit)| if emit { None } else { Some(path) })
            .collect();

        EntityBytecode {
            state_id: self.state_id,
            handlers,
            entity_name: self.entity_name.clone(),
            when_events,
            non_emitted_fields,
            computed_paths: self.spec.computed_fields.clone(),
            computed_fields_evaluator: None,
        }
    }

    /// The sibling instruction types this entity's registrations read, each
    /// with the fields read from it, in a deterministic order.
    fn sibling_capture_paths(&self) -> Vec<(String, Vec<FieldPath>)> {
        let mut captures: std::collections::BTreeMap<String, Vec<FieldPath>> =
            std::collections::BTreeMap::new();
        for hook in &self.spec.instruction_hooks {
            for action in &hook.actions {
                let HookAction::RegisterPdaMapping {
                    pda_field,
                    seed_field,
                    sibling: Some(sibling),
                    ..
                } = action
                else {
                    continue;
                };
                let paths = captures
                    .entry(sibling.instruction_type.clone())
                    .or_default();
                for path in sibling.required_fields.keys() {
                    let path = FieldPath::new(&path.split('.').collect::<Vec<_>>());
                    if !paths.contains(&path) {
                        paths.push(path);
                    }
                }
                for (from_sibling, path) in [
                    (sibling.pda_from_sibling, pda_field),
                    (sibling.seed_from_sibling, seed_field),
                ] {
                    if from_sibling && !paths.iter().any(|known| known.segments == path.segments) {
                        paths.push(path.clone());
                    }
                }
            }
        }
        captures.into_iter().collect()
    }

    /// Canonical encoding of the opcodes that resolve a handler's key. Two
    /// handlers for the same event may share one state read/write only when
    /// these are identical.
    fn key_identity(&self, spec: &TypedHandlerSpec<S>) -> Value {
        let key_reg = 20;
        let ops = self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings);
        serde_json::to_value(&ops).unwrap_or(Value::Null)
    }

    /// How a hook's `lookup_by` field resolves the entity key: directly when
    /// it names a primary-key field, through its lookup index when it names a
    /// lookup-index field.
    fn hook_key_resolution(&self, lookup_by: &FieldPath) -> Option<KeyResolutionStrategy> {
        let leaf = lookup_by.segments.last()?;
        let is_primary_key_field = self
            .spec
            .identity
            .primary_keys
            .iter()
            .any(|pk| pk.rsplit('.').next() == Some(leaf.as_str()));
        if is_primary_key_field {
            Some(KeyResolutionStrategy::Embedded {
                primary_field: lookup_by.clone(),
            })
        } else if self.find_lookup_index_for_field(lookup_by).is_some() {
            Some(KeyResolutionStrategy::Lookup {
                primary_field: lookup_by.clone(),
                index_name: None,
            })
        } else {
            None
        }
    }

    /// Segment for an instruction hook that routes by its own resolved key:
    /// key loading, state read and state write. Hook actions are inserted
    /// before the state write by the caller.
    fn compile_keyed_hook_segment(&self, resolution: &KeyResolutionStrategy) -> Vec<OpCode> {
        let key_reg = 20;
        let state_reg = 2;
        let mut ops = self.compile_key_loading(resolution, key_reg, &[]);
        ops.push(OpCode::AbortIfNullKey {
            key: key_reg,
            is_account_event: false,
        });
        ops.push(OpCode::ReadOrInitState {
            state_id: self.state_id,
            key: key_reg,
            default: serde_json::json!({}),
            dest: state_reg,
            dedup_route: None,
        });
        ops.push(OpCode::UpdateState {
            state_id: self.state_id,
            key: key_reg,
            value: state_reg,
        });
        ops
    }

    /// Standalone segment for an instruction hook: key from the resolver or
    /// the hook's `lookup_by` field, then state read and write. Hook actions are
    /// inserted before the state write by the caller.
    fn compile_hook_segment(&self, hook: &InstructionHook) -> Vec<OpCode> {
        let key_reg = 20;
        let state_reg = 2;
        let resolved_key_reg = 19;
        let temp_reg = 18;

        let mut ops = Vec::new();

        // First, try to load __resolved_primary_key from resolver
        ops.push(OpCode::LoadEventField {
            path: FieldPath::new(&["__resolved_primary_key"]),
            dest: resolved_key_reg,
            default: Some(serde_json::json!(null)),
        });

        // Copy to key_reg (unconditionally, may be null)
        ops.push(OpCode::CopyRegister {
            source: resolved_key_reg,
            dest: key_reg,
        });

        // If hook has lookup_by, use it to load primary key from instruction accounts
        if let Some(lookup_path) = &hook.lookup_by {
            // Load the primary key from the instruction's lookup_by field (e.g., accounts.signer)
            ops.push(OpCode::LoadEventField {
                path: lookup_path.clone(),
                dest: temp_reg,
                default: None,
            });

            // Apply HexEncode transformation (accounts are byte arrays)
            ops.push(OpCode::Transform {
                source: temp_reg,
                dest: temp_reg,
                transformation: Transformation::HexEncode,
            });

            // Use this as fallback if __resolved_primary_key was null
            ops.push(OpCode::CopyRegisterIfNull {
                source: temp_reg,
                dest: key_reg,
            });
        }

        ops.push(OpCode::ReadOrInitState {
            state_id: self.state_id,
            key: key_reg,
            default: serde_json::json!({}),
            dest: state_reg,
            dedup_route: None,
        });

        ops.push(OpCode::UpdateState {
            state_id: self.state_id,
            key: key_reg,
            value: state_reg,
        });

        ops
    }

    fn compile_handler(&self, spec: &TypedHandlerSpec<S>) -> Vec<OpCode> {
        let mut ops = Vec::new();
        let state_reg = 2;
        let key_reg = 20;

        ops.extend(self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings));

        // Guard: if key resolved to null on an account-state event, abort
        // early with empty mutations so process_event can queue the update
        // for later reprocessing.  Without this, downstream opcodes would
        // create a phantom entity keyed by null and produce non-empty
        // mutations that prevent queueing.
        let is_account_event = matches!(
            spec.source,
            SourceSpec::Source {
                is_account: true,
                ..
            }
        );
        ops.push(OpCode::AbortIfNullKey {
            key: key_reg,
            is_account_event,
        });

        ops.push(OpCode::ReadOrInitState {
            state_id: self.state_id,
            key: key_reg,
            default: serde_json::json!({}),
            dest: state_reg,
            dedup_route: declared_index(&spec.key_resolution)
                .map(|index| format!("{index}:{:?}", spec.key_resolution)),
        });

        // Index updates must come AFTER ReadOrInitState so the state table exists.
        // ReadOrInitState lazily creates the state table via entry().or_insert_with(),
        // but index opcodes (UpdateLookupIndex, UpdateTemporalIndex, UpdatePdaReverseLookup)
        // use get_mut() which fails if the table doesn't exist yet.
        // This ordering also means stale/duplicate updates (caught by ReadOrInitState's
        // recency check) correctly skip index updates too.
        ops.extend(self.compile_temporal_index_update(
            &spec.key_resolution,
            key_reg,
            &spec.mappings,
        ));

        for mapping in &spec.mappings {
            ops.extend(self.compile_mapping(mapping, state_reg, key_reg));
        }

        ops.extend(self.compile_resolvers(state_reg, key_reg));

        // Evaluate computed fields after all mappings but before updating state
        ops.push(OpCode::EvaluateComputedFields {
            state: state_reg,
            computed_paths: self.spec.computed_fields.clone(),
        });

        ops.push(OpCode::UpdateState {
            state_id: self.state_id,
            key: key_reg,
            value: state_reg,
        });

        if spec.emit {
            ops.push(OpCode::EmitMutation {
                entity_name: self.entity_name.clone(),
                key: key_reg,
                state: state_reg,
            });
        }

        ops
    }

    fn compile_resolvers(&self, state_reg: Register, key_reg: Register) -> Vec<OpCode> {
        let mut ops = Vec::new();

        for resolver_spec in &self.spec.resolver_specs {
            let url_template = match &resolver_spec.resolver {
                ResolverType::Url(config) => match &config.url_source {
                    UrlSource::Template(parts) => Some(parts.clone()),
                    _ => None,
                },
                _ => None,
            };

            ops.push(OpCode::QueueResolver {
                state_id: self.state_id,
                entity_name: self.entity_name.clone(),
                resolver: resolver_spec.resolver.clone(),
                input_path: resolver_spec.input_path.clone(),
                input_value: resolver_spec.input_value.clone(),
                url_template,
                strategy: resolver_spec.strategy.clone(),
                extracts: resolver_spec.extracts.clone(),
                condition: resolver_spec.condition.clone(),
                schedule_at: resolver_spec.schedule_at.clone(),
                state: state_reg,
                key: key_reg,
            });
        }

        ops
    }

    fn compile_mapping(
        &self,
        mapping: &TypedFieldMapping<S>,
        state_reg: Register,
        key_reg: Register,
    ) -> Vec<OpCode> {
        let guard = mapping
            .condition
            .as_ref()
            .and_then(|condition| condition.parsed.as_ref())
            .and_then(|parsed| match parsed {
                ParsedCondition::Guard { expr } => Some(expr),
                _ => None,
            });
        let Some(guard) = guard else {
            return self.compile_mapping_ops(
                mapping,
                mapping.condition.as_ref(),
                state_reg,
                key_reg,
            );
        };
        // A guarded mapping is the unconditional mapping, skipped as a whole
        // unless the guard over the incoming update is true.
        let inner = self.compile_mapping_ops(mapping, None, state_reg, key_reg);
        let mut ops = Vec::with_capacity(inner.len() + 2);
        ops.push(OpCode::EvaluateEventExpr {
            expr: guard.clone(),
            transform: None,
            dest: GUARD_REG,
        });
        ops.push(OpCode::SkipUnless {
            predicate: GUARD_REG,
            count: inner.len(),
        });
        ops.extend(inner);
        ops
    }

    fn compile_mapping_ops(
        &self,
        mapping: &TypedFieldMapping<S>,
        condition: Option<&ConditionExpr>,
        state_reg: Register,
        key_reg: Register,
    ) -> Vec<OpCode> {
        let mut ops = Vec::new();
        let temp_reg = 10;

        ops.extend(self.compile_mapping_source(&mapping.source, temp_reg));

        if let Some(transform) = &mapping.transform {
            ops.push(OpCode::Transform {
                source: temp_reg,
                dest: temp_reg,
                transformation: transform.clone(),
            });
        }

        if let Some(stop_instruction) = &mapping.stop {
            if mapping.when.is_some() {
                tracing::warn!(
                    "#[map] stop and when both set for {}. Ignoring when.",
                    mapping.target_path
                );
            }
            if !matches!(mapping.population, PopulationStrategy::LastWrite)
                && !matches!(mapping.population, PopulationStrategy::Merge)
            {
                tracing::warn!(
                    "#[map] stop ignores population strategy {:?}",
                    mapping.population
                );
            }

            ops.push(OpCode::SetFieldUnlessStopped {
                object: state_reg,
                path: mapping.target_path.clone(),
                value: temp_reg,
                stop_field: stop_field_path(&mapping.target_path),
                stop_instruction: stop_instruction.clone(),
                entity_name: self.entity_name.clone(),
                key_reg,
            });
            return ops;
        }

        if let Some(when_instruction) = &mapping.when {
            if !matches!(mapping.population, PopulationStrategy::LastWrite)
                && !matches!(mapping.population, PopulationStrategy::Merge)
            {
                tracing::warn!(
                    "#[map] when ignores population strategy {:?}",
                    mapping.population
                );
            }
            let (condition_field, condition_op, condition_value) = condition
                .and_then(|cond| cond.parsed.as_ref())
                .and_then(|parsed| match parsed {
                    ParsedCondition::Comparison { field, op, value } => {
                        Some((Some(field.clone()), Some(op.clone()), Some(value.clone())))
                    }
                    ParsedCondition::Logical { .. } => {
                        tracing::warn!("Logical conditions not yet supported for #[map] when");
                        None
                    }
                    // compile_mapping evaluates guards before this point.
                    ParsedCondition::Guard { .. } => None,
                })
                .unwrap_or((None, None, None));

            ops.push(OpCode::SetFieldWhen {
                object: state_reg,
                path: mapping.target_path.clone(),
                value: temp_reg,
                when_instruction: when_instruction.clone(),
                entity_name: self.entity_name.clone(),
                key_reg,
                condition_field,
                condition_op,
                condition_value,
            });
            return ops;
        }

        if let Some(condition) = condition {
            if let Some(parsed) = &condition.parsed {
                match parsed {
                    ParsedCondition::Comparison {
                        field,
                        op,
                        value: cond_value,
                    } => {
                        if matches!(mapping.population, PopulationStrategy::LastWrite)
                            || matches!(mapping.population, PopulationStrategy::Merge)
                        {
                            ops.push(OpCode::ConditionalSetField {
                                object: state_reg,
                                path: mapping.target_path.clone(),
                                value: temp_reg,
                                condition_field: field.clone(),
                                condition_op: op.clone(),
                                condition_value: cond_value.clone(),
                            });
                            return ops;
                        }

                        if matches!(mapping.population, PopulationStrategy::Count) {
                            ops.push(OpCode::ConditionalIncrement {
                                object: state_reg,
                                path: mapping.target_path.clone(),
                                condition_field: field.clone(),
                                condition_op: op.clone(),
                                condition_value: cond_value.clone(),
                            });
                            return ops;
                        }

                        tracing::warn!(
                            "Conditional #[map] not supported for population strategy {:?}",
                            mapping.population
                        );
                    }
                    ParsedCondition::Logical { .. } => {
                        tracing::warn!("Logical conditions not yet supported for #[map]");
                    }
                    // compile_mapping evaluates guards before this point.
                    ParsedCondition::Guard { .. } => {}
                }
            }
        }

        match &mapping.population {
            PopulationStrategy::Append => {
                ops.push(OpCode::AppendToArray {
                    object: state_reg,
                    path: mapping.target_path.clone(),
                    value: temp_reg,
                });
            }
            PopulationStrategy::LastWrite => {
                ops.push(OpCode::SetField {
                    object: state_reg,
                    path: mapping.target_path.clone(),
                    value: temp_reg,
                });
            }
            PopulationStrategy::SetOnce => {
                ops.push(OpCode::SetFieldIfNull {
                    object: state_reg,
                    path: mapping.target_path.clone(),
                    value: temp_reg,
                });
            }
            PopulationStrategy::Merge => {
                ops.push(OpCode::SetField {
                    object: state_reg,
                    path: mapping.target_path.clone(),
                    value: temp_reg,
                });
            }
            PopulationStrategy::Max => {
                ops.push(OpCode::SetFieldMax {
                    object: state_reg,
                    path: mapping.target_path.clone(),
                    value: temp_reg,
                });
            }
            PopulationStrategy::Sum => {
                ops.push(OpCode::SetFieldSum {
                    object: state_reg,
                    path: mapping.target_path.clone(),
                    value: temp_reg,
                });
            }
            PopulationStrategy::Count => {
                // Count doesn't need the value, just increment
                ops.push(OpCode::SetFieldIncrement {
                    object: state_reg,
                    path: mapping.target_path.clone(),
                });
            }
            PopulationStrategy::Min => {
                ops.push(OpCode::SetFieldMin {
                    object: state_reg,
                    path: mapping.target_path.clone(),
                    value: temp_reg,
                });
            }
            PopulationStrategy::UniqueCount => {
                // UniqueCount requires maintaining an internal set
                // The field stores the count, but we track unique values in a hidden set
                let set_name = format!("{}_unique_set", mapping.target_path);
                ops.push(OpCode::AddToUniqueSet {
                    state_id: self.state_id,
                    set_name,
                    value: temp_reg,
                    count_object: state_reg,
                    count_path: mapping.target_path.clone(),
                });
            }
        }

        ops
    }

    fn compile_mapping_source(&self, source: &MappingSource, dest: Register) -> Vec<OpCode> {
        match source {
            MappingSource::FromSource {
                path,
                default,
                transform,
            } => {
                let mut ops = vec![OpCode::LoadEventField {
                    path: path.clone(),
                    dest,
                    default: default.clone(),
                }];

                // Apply transform if specified in the source
                if let Some(transform_type) = transform {
                    ops.push(OpCode::Transform {
                        source: dest,
                        dest,
                        transformation: transform_type.clone(),
                    });
                }

                ops
            }
            MappingSource::Constant(val) => {
                vec![OpCode::LoadConstant {
                    value: val.clone(),
                    dest,
                }]
            }
            MappingSource::AsEvent { fields } => {
                let mut ops = Vec::new();

                if fields.is_empty() {
                    let event_data_reg = dest + 1;
                    ops.push(OpCode::LoadEventField {
                        path: FieldPath::new(&[]),
                        dest: event_data_reg,
                        default: Some(serde_json::json!({})),
                    });
                    ops.push(OpCode::CreateEvent {
                        dest,
                        event_value: event_data_reg,
                    });
                } else {
                    let data_obj_reg = dest + 1;
                    ops.push(OpCode::CreateObject { dest: data_obj_reg });

                    let mut field_registers = Vec::new();
                    let mut current_reg = dest + 2;

                    for field_source in fields.iter() {
                        if let MappingSource::FromSource {
                            path,
                            default,
                            transform,
                        } = &**field_source
                        {
                            ops.push(OpCode::LoadEventField {
                                path: path.clone(),
                                dest: current_reg,
                                default: default.clone(),
                            });

                            if let Some(transform_type) = transform {
                                ops.push(OpCode::Transform {
                                    source: current_reg,
                                    dest: current_reg,
                                    transformation: transform_type.clone(),
                                });
                            }

                            if let Some(field_name) = path.segments.last() {
                                field_registers.push((field_name.clone(), current_reg));
                            }
                            current_reg += 1;
                        }
                    }

                    if !field_registers.is_empty() {
                        ops.push(OpCode::SetFields {
                            object: data_obj_reg,
                            fields: field_registers,
                        });
                    }

                    ops.push(OpCode::CreateEvent {
                        dest,
                        event_value: data_obj_reg,
                    });
                }

                ops
            }
            MappingSource::WholeSource => {
                vec![OpCode::LoadEventField {
                    path: FieldPath::new(&[]),
                    dest,
                    default: Some(serde_json::json!({})),
                }]
            }
            MappingSource::AsCapture { field_transforms } => {
                // AsCapture loads the whole source, applies field-level transforms, and wraps in CaptureWrapper
                let capture_data_reg = 22; // Temp register for capture data before wrapping
                let mut ops = vec![OpCode::LoadEventField {
                    path: FieldPath::new(&[]),
                    dest: capture_data_reg,
                    default: Some(serde_json::json!({})),
                }];

                // Apply transforms to specific fields in the loaded object
                // IMPORTANT: Use registers that don't conflict with key_reg (20)
                // Using 24 and 25 to avoid conflicts with key loading (uses 18, 19, 20, 23)
                let field_reg = 24;
                let transformed_reg = 25;

                for (field_name, transform) in field_transforms {
                    // Load the field from the capture_data_reg (not from event!)
                    // Use GetField opcode to read from a register instead of LoadEventField
                    ops.push(OpCode::GetField {
                        object: capture_data_reg,
                        path: field_name.clone(),
                        dest: field_reg,
                    });

                    // Transform it
                    ops.push(OpCode::Transform {
                        source: field_reg,
                        dest: transformed_reg,
                        transformation: transform.clone(),
                    });

                    // Set it back into the capture data object
                    ops.push(OpCode::SetField {
                        object: capture_data_reg,
                        path: field_name.clone(),
                        value: transformed_reg,
                    });
                }

                // Wrap the capture data in CaptureWrapper with metadata
                ops.push(OpCode::CreateCapture {
                    dest,
                    capture_value: capture_data_reg,
                });

                ops
            }
            MappingSource::FromContext { field } => {
                // Load from instruction context (timestamp, slot, signature)
                vec![OpCode::LoadEventField {
                    path: FieldPath::new(&["__update_context", field.as_str()]),
                    dest,
                    default: Some(serde_json::json!(null)),
                }]
            }
            MappingSource::FromExpr { expr } => {
                vec![OpCode::EvaluateEventExpr {
                    expr: expr.clone(),
                    transform: None,
                    dest,
                }]
            }
            MappingSource::Computed { .. } => {
                vec![]
            }
            MappingSource::FromState { .. } => {
                vec![]
            }
        }
    }

    pub fn compile_key_loading(
        &self,
        resolution: &KeyResolutionStrategy,
        key_reg: Register,
        mappings: &[TypedFieldMapping<S>],
    ) -> Vec<OpCode> {
        let mut ops = Vec::new();

        // Resolvers provide either a final key or an intermediate lookup key,
        // depending on the handler strategy.
        let resolved_key_reg = 19; // Use a temp register
        ops.push(OpCode::LoadEventField {
            path: FieldPath::new(&["__resolved_primary_key"]),
            dest: resolved_key_reg,
            default: Some(serde_json::json!(null)),
        });

        // Now do the normal key resolution
        match resolution {
            KeyResolutionStrategy::Embedded { primary_field }
                if primary_field.segments.is_empty()
                    && self.find_primary_key_expr(mappings).is_some() =>
            {
                // The primary key mapping is an expression over the event,
                // such as a pool and a time bucket.
                let expr = self
                    .find_primary_key_expr(mappings)
                    .expect("checked by the guard");
                // A write guard on the key mapping gates the key too: when it
                // fails the key stays null and the update writes no row.
                if let Some(guard) = self.find_primary_key_guard(mappings) {
                    ops.push(OpCode::EvaluateEventExpr {
                        expr: guard,
                        transform: None,
                        dest: GUARD_REG,
                    });
                    ops.push(OpCode::SkipUnless {
                        predicate: GUARD_REG,
                        count: 1,
                    });
                }
                // The key is the value the mapping writes to the primary key
                // field, transform included; a null key stays null, so the
                // event writes no row.
                ops.push(OpCode::EvaluateEventExpr {
                    expr,
                    transform: self.find_primary_key_transformation(mappings),
                    dest: key_reg,
                });
                ops.push(OpCode::CopyRegisterIfNull {
                    source: resolved_key_reg,
                    dest: key_reg,
                });
            }
            KeyResolutionStrategy::Embedded { primary_field } => {
                // Enhanced key resolution: check for auto-inheritance when primary_field is empty
                let effective_primary_field = if primary_field.segments.is_empty() {
                    // Try to auto-detect primary field from account schema
                    if let Some(auto_field) = self.auto_detect_primary_field(mappings) {
                        auto_field
                    } else {
                        primary_field.clone()
                    }
                } else {
                    primary_field.clone()
                };

                // Skip fallback key loading if effective primary_field is still empty
                // This happens for account types that rely solely on __resolved_primary_key
                // (e.g., accounts with #[resolve_key_for] resolvers)
                if !effective_primary_field.segments.is_empty() {
                    let temp_reg = 18;
                    let transform_reg = 23; // Register for transformed key

                    ops.push(OpCode::LoadEventField {
                        path: effective_primary_field.clone(),
                        dest: temp_reg,
                        default: None,
                    });

                    // Check if there's a transformation for the primary key field
                    // First try the current mappings, then inherited transformations
                    let primary_key_transform = self
                        .find_primary_key_transformation(mappings)
                        .or_else(|| self.find_inherited_primary_key_transformation());

                    if let Some(transform) = primary_key_transform {
                        // Apply transformation to the loaded key
                        ops.push(OpCode::Transform {
                            source: temp_reg,
                            dest: transform_reg,
                            transformation: transform,
                        });
                        // A concrete embedded key belongs to this entity and
                        // must win over a resolver result injected for another
                        // handler consuming the same event.
                        ops.push(OpCode::CopyRegister {
                            source: transform_reg,
                            dest: key_reg,
                        });
                    } else {
                        // No transformation, use raw value.
                        ops.push(OpCode::CopyRegister {
                            source: temp_reg,
                            dest: key_reg,
                        });
                    }
                    ops.push(OpCode::CopyRegisterIfNull {
                        source: resolved_key_reg,
                        dest: key_reg,
                    });
                } else {
                    // Resolver-only embedded handlers have no local key field.
                    ops.push(OpCode::CopyRegister {
                        source: resolved_key_reg,
                        dest: key_reg,
                    });
                }
            }
            KeyResolutionStrategy::Lookup {
                primary_field,
                index_name: declared,
            } => {
                let lookup_reg = 15;
                let result_reg = 17;

                // Prefer resolver-provided key as lookup input.
                // When __resolved_primary_key is set (e.g. round_address from
                // PDA reverse lookup), use it directly — this gives a one-hop
                // lookup (round_address → round_id) instead of a two-hop chain
                // (Var address → PDA → round_address → round_id).
                ops.push(OpCode::CopyRegister {
                    source: resolved_key_reg,
                    dest: lookup_reg,
                });

                let temp_reg = 18;
                ops.push(OpCode::LoadEventField {
                    path: primary_field.clone(),
                    dest: temp_reg,
                    default: None,
                });
                ops.push(OpCode::CopyRegisterIfNull {
                    source: temp_reg,
                    dest: lookup_reg,
                });

                // A declared index is used as is, and strictly; otherwise
                // the index is inferred from the field names, and the
                // runtime falls back to the entity's other indexes.
                let effective_index_name = match declared {
                    Some(declared) => declared.clone(),
                    None => self
                        .find_lookup_index_for_lookup_field(primary_field, mappings)
                        .unwrap_or_else(|| "default_pda_lookup".to_string()),
                };

                ops.push(OpCode::LookupIndex {
                    state_id: self.state_id,
                    index_name: effective_index_name,
                    lookup_value: lookup_reg,
                    dest: result_reg,
                    strict: declared.is_some(),
                });
                ops.push(OpCode::CopyRegister {
                    source: result_reg,
                    dest: key_reg,
                });

                // Most lookup-based account handlers expect resolver output to be an
                // intermediate lookup value (for example, PDA -> round_address -> round_id),
                // so a null LookupIndex result must leave the key null for queueing.
                // Some stacks register the PDA directly to the entity primary key
                // (for example, bonding_curve -> mint). In that case the resolver output
                // is already the final key, so preserve it only when the instruction hook's
                // seed field matches one of the entity primary key fields.
                if self.resolver_outputs_primary_key_directly(primary_field) {
                    ops.push(OpCode::CopyRegisterIfNull {
                        source: resolved_key_reg,
                        dest: key_reg,
                    });
                }
            }
            KeyResolutionStrategy::Computed {
                primary_field,
                compute_partition: _,
            } => {
                // Copy resolver result to key_reg (may be null)
                ops.push(OpCode::CopyRegister {
                    source: resolved_key_reg,
                    dest: key_reg,
                });
                let temp_reg = 18;
                ops.push(OpCode::LoadEventField {
                    path: primary_field.clone(),
                    dest: temp_reg,
                    default: None,
                });
                ops.push(OpCode::CopyRegisterIfNull {
                    source: temp_reg,
                    dest: key_reg,
                });
            }
            KeyResolutionStrategy::TemporalLookup {
                lookup_field,
                timestamp_field,
                index_name,
            } => {
                // Copy resolver result to key_reg (may be null)
                ops.push(OpCode::CopyRegister {
                    source: resolved_key_reg,
                    dest: key_reg,
                });
                let lookup_reg = 15;
                let timestamp_reg = 16;
                let result_reg = 17;

                ops.push(OpCode::LoadEventField {
                    path: lookup_field.clone(),
                    dest: lookup_reg,
                    default: None,
                });

                ops.push(OpCode::LoadEventField {
                    path: timestamp_field.clone(),
                    dest: timestamp_reg,
                    default: None,
                });

                ops.push(OpCode::LookupTemporalIndex {
                    state_id: self.state_id,
                    index_name: index_name.clone(),
                    lookup_value: lookup_reg,
                    timestamp: timestamp_reg,
                    dest: result_reg,
                });

                ops.push(OpCode::CopyRegisterIfNull {
                    source: result_reg,
                    dest: key_reg,
                });
            }
        }

        ops
    }

    /// The expression of a mapping in `mappings` that sets the primary key,
    /// if one does.
    fn find_primary_key_expr(&self, mappings: &[TypedFieldMapping<S>]) -> Option<ComputedExpr> {
        let primary_key = self.spec.identity.primary_keys.first()?;
        mappings.iter().find_map(|mapping| match &mapping.source {
            MappingSource::FromExpr { expr } if mapping.target_path == *primary_key => {
                Some(expr.clone())
            }
            _ => None,
        })
    }

    /// The write guard of the expression mapping that sets the primary key,
    /// if it has one.
    fn find_primary_key_guard(&self, mappings: &[TypedFieldMapping<S>]) -> Option<ComputedExpr> {
        let primary_key = self.spec.identity.primary_keys.first()?;
        mappings.iter().find_map(|mapping| {
            if mapping.target_path != *primary_key
                || !matches!(mapping.source, MappingSource::FromExpr { .. })
            {
                return None;
            }
            match mapping.condition.as_ref()?.parsed.as_ref()? {
                ParsedCondition::Guard { expr } => Some(expr.clone()),
                _ => None,
            }
        })
    }

    fn find_primary_key_transformation(
        &self,
        mappings: &[TypedFieldMapping<S>],
    ) -> Option<Transformation> {
        // Find the first primary key in the identity spec
        let primary_key = self.spec.identity.primary_keys.first()?;
        let primary_field_name = self.extract_primary_field_name(primary_key)?;

        // Look for a mapping that targets this primary key
        for mapping in mappings {
            // Check if this mapping targets the primary key field
            if mapping.target_path == *primary_key
                || mapping.target_path.ends_with(&format!(".{}", primary_key))
            {
                // Check mapping-level transform first
                if let Some(transform) = &mapping.transform {
                    return Some(transform.clone());
                }

                // Then check source-level transform
                if let MappingSource::FromSource {
                    transform: Some(transform),
                    ..
                } = &mapping.source
                {
                    return Some(transform.clone());
                }
            }
        }

        // If no explicit primary key mapping found, check AsCapture field transforms
        for mapping in mappings {
            if let MappingSource::AsCapture { field_transforms } = &mapping.source {
                if let Some(transform) = field_transforms.get(&primary_field_name) {
                    return Some(transform.clone());
                }
            }
        }

        None
    }

    /// Look for primary key mappings in other handlers of the same entity
    /// This enables cross-handler inheritance of key transformations
    pub fn find_inherited_primary_key_transformation(&self) -> Option<Transformation> {
        let primary_key = self.spec.identity.primary_keys.first()?;

        // Extract the field name from the primary key path (e.g., "id.authority" -> "authority")
        let primary_field_name = self.extract_primary_field_name(primary_key)?;

        // Search through all handlers in the spec for primary key mappings
        for handler in &self.spec.handlers {
            for mapping in &handler.mappings {
                // Look for mappings targeting the primary key
                if mapping.target_path == *primary_key
                    || mapping.target_path.ends_with(&format!(".{}", primary_key))
                {
                    // Check if this mapping comes from a field matching the primary key name
                    if let MappingSource::FromSource {
                        path, transform, ..
                    } = &mapping.source
                    {
                        if path.segments.last() == Some(&primary_field_name) {
                            // Return mapping-level transform first, then source-level transform
                            return mapping.transform.clone().or_else(|| transform.clone());
                        }
                    }
                }

                // Also check AsCapture field transforms for the primary field
                if let MappingSource::AsCapture { field_transforms } = &mapping.source {
                    if let Some(transform) = field_transforms.get(&primary_field_name) {
                        return Some(transform.clone());
                    }
                }
            }
        }

        None
    }

    /// Extract the field name from a primary key path (e.g., "id.authority" -> "authority")
    fn extract_primary_field_name(&self, primary_key: &str) -> Option<String> {
        // Split by '.' and take the last segment
        primary_key.split('.').next_back().map(|s| s.to_string())
    }

    /// Auto-detect primary field from account schema when no explicit mapping exists
    /// This looks for account types that have an 'authority' field and tries to use it
    pub fn auto_detect_primary_field(
        &self,
        current_mappings: &[TypedFieldMapping<S>],
    ) -> Option<FieldPath> {
        let primary_key = self.spec.identity.primary_keys.first()?;

        // Extract the field name from the primary key (e.g., "id.authority" -> "authority")
        let primary_field_name = self.extract_primary_field_name(primary_key)?;

        // Check if current handler can access the primary field
        if self.current_account_has_primary_field(&primary_field_name, current_mappings) {
            return Some(FieldPath::new(&[&primary_field_name]));
        }

        None
    }

    /// Check if the current account type has the primary field
    /// This is determined by looking at the mappings to see what fields are available
    fn current_account_has_primary_field(
        &self,
        field_name: &str,
        mappings: &[TypedFieldMapping<S>],
    ) -> bool {
        // Look through the mappings to see if any reference the primary field
        for mapping in mappings {
            if let MappingSource::FromSource { path, .. } = &mapping.source {
                // Check if this mapping sources from the primary field
                if path.segments.last() == Some(&field_name.to_string()) {
                    return true;
                }
            }
        }

        false
    }

    /// Check if handler has access to a specific field in its source account
    #[allow(dead_code)]
    fn handler_has_field(&self, field_name: &str, mappings: &[TypedFieldMapping<S>]) -> bool {
        for mapping in mappings {
            if let MappingSource::FromSource { path, .. } = &mapping.source {
                if path.segments.last() == Some(&field_name.to_string()) {
                    return true;
                }
            }
        }
        false
    }

    /// Check if field exists by looking at mappings (IDL-agnostic approach)
    /// This avoids hardcoding account schemas and uses actual mapping evidence
    #[allow(dead_code)]
    fn field_exists_in_mappings(
        &self,
        field_name: &str,
        mappings: &[TypedFieldMapping<S>],
    ) -> bool {
        // Look through current mappings to see if the field is referenced
        for mapping in mappings {
            if let MappingSource::FromSource { path, .. } = &mapping.source {
                if path.segments.last() == Some(&field_name.to_string()) {
                    return true;
                }
            }
            // Also check AsCapture field transforms
            if let MappingSource::AsCapture { field_transforms } = &mapping.source {
                if field_transforms.contains_key(field_name) {
                    return true;
                }
            }
        }
        false
    }

    fn find_lookup_index_for_field(&self, field_path: &FieldPath) -> Option<String> {
        if field_path.segments.is_empty() {
            return None;
        }

        let lookup_field_name = field_path.segments.last().unwrap();

        for lookup_index in &self.spec.identity.lookup_indexes {
            let index_field_name = lookup_index
                .field_name
                .split('.')
                .next_back()
                .unwrap_or(&lookup_index.field_name);
            let matches_directly = index_field_name == lookup_field_name;
            // An index field named `foo_address` is treated as an alias for the
            // bare field `foo` (for example, `mint_address` resolves handlers
            // keyed on `mint`). This is intentionally one-way: bare `foo` does
            // not imply a `foo_address` lookup index. The macro crate mirrors
            // this convention via `lookup_index_leafs` in
            // `arete-macros/src/validation/mod.rs`.
            let matches_address_alias = index_field_name
                .strip_suffix("_address")
                .map(|base| base == lookup_field_name)
                .unwrap_or(false);

            if matches_directly || matches_address_alias {
                return Some(format!("{}_lookup_index", index_field_name));
            }
        }

        None
    }

    /// Whether a handler of this entity declares it resolves through
    /// `index_name` (a declared, strict lookup).
    fn declares_index(&self, index_name: &str) -> bool {
        self.spec
            .handlers
            .iter()
            .any(|handler| declared_index(&handler.key_resolution) == Some(index_name))
    }

    /// Find lookup index for a Lookup key resolution by checking if there's a mapping
    /// from the primary_field to a lookup index field.
    fn find_lookup_index_for_lookup_field(
        &self,
        primary_field: &FieldPath,
        mappings: &[TypedFieldMapping<S>],
    ) -> Option<String> {
        // Build the primary field path string
        let primary_path = primary_field.segments.join(".");

        // Check if there's a mapping from this primary field to a lookup index field
        for mapping in mappings {
            // Check if the mapping source path matches the primary field
            if let MappingSource::FromSource { path, .. } = &mapping.source {
                let source_path = path.segments.join(".");
                if source_path == primary_path {
                    // Check if the target is a lookup index field
                    for lookup_index in &self.spec.identity.lookup_indexes {
                        if mapping.target_path == lookup_index.field_name {
                            let index_field_name = lookup_index
                                .field_name
                                .split('.')
                                .next_back()
                                .unwrap_or(&lookup_index.field_name);
                            return Some(format!("{}_lookup_index", index_field_name));
                        }
                    }
                }
            }
        }

        // Fall back to direct field name matching
        self.find_lookup_index_for_field(primary_field)
    }

    /// Find the source path for a lookup index field by looking at mappings.
    /// For example, if target_path is "id.round_address" and the mapping is
    /// `id.round_address <- __account_address`, this returns ["__account_address"].
    fn find_source_path_for_lookup_index(
        &self,
        mappings: &[TypedFieldMapping<S>],
        lookup_field_name: &str,
    ) -> Option<Vec<String>> {
        for mapping in mappings {
            if mapping.target_path == lookup_field_name {
                if let MappingSource::FromSource { path, .. } = &mapping.source {
                    return Some(path.segments.clone());
                }
            }
        }
        None
    }

    fn compile_temporal_index_update(
        &self,
        resolution: &KeyResolutionStrategy,
        key_reg: Register,
        mappings: &[TypedFieldMapping<S>],
    ) -> Vec<OpCode> {
        let mut ops = Vec::new();

        for lookup_index in &self.spec.identity.lookup_indexes {
            let lookup_reg = 17;
            let source_field = lookup_index
                .field_name
                .split('.')
                .next_back()
                .unwrap_or(&lookup_index.field_name);

            let start = ops.len();
            match resolution {
                KeyResolutionStrategy::Embedded { primary_field: _ } => {
                    // For Embedded handlers, find the mapping that targets this lookup index field
                    // and use its source path to load the lookup value
                    let source_path_opt =
                        self.find_source_path_for_lookup_index(mappings, &lookup_index.field_name);

                    let load_path = if let Some(ref path) = source_path_opt {
                        FieldPath::new(&path.iter().map(|s| s.as_str()).collect::<Vec<_>>())
                    } else {
                        // Fallback to source_field if no mapping found
                        FieldPath::new(&[source_field])
                    };

                    if let Some(mapping) = mappings
                        .iter()
                        .find(|m| m.target_path == lookup_index.field_name)
                    {
                        let value_ops = self.compile_mapping_source(&mapping.source, lookup_reg);
                        if value_ops.is_empty() {
                            continue;
                        }
                        ops.extend(value_ops);
                        if let Some(transform) = &mapping.transform {
                            ops.push(OpCode::Transform {
                                source: lookup_reg,
                                dest: lookup_reg,
                                transformation: transform.clone(),
                            });
                        }
                    } else {
                        ops.push(OpCode::LoadEventField {
                            path: load_path,
                            dest: lookup_reg,
                            default: None,
                        });
                    }

                    if let Some(temporal_field_name) = &lookup_index.temporal_field {
                        let timestamp_reg = 18;

                        ops.push(OpCode::LoadEventField {
                            path: FieldPath::new(&[temporal_field_name]),
                            dest: timestamp_reg,
                            default: None,
                        });

                        let index_name = format!("{}_temporal_index", source_field);
                        ops.push(OpCode::UpdateTemporalIndex {
                            state_id: self.state_id,
                            index_name,
                            lookup_value: lookup_reg,
                            primary_key: key_reg,
                            timestamp: timestamp_reg,
                        });

                        let simple_index_name = format!("{}_lookup_index", source_field);
                        ops.push(OpCode::UpdateLookupIndex {
                            state_id: self.state_id,
                            index_name: simple_index_name,
                            lookup_value: lookup_reg,
                            primary_key: key_reg,
                        });
                    } else {
                        let index_name = format!("{}_lookup_index", source_field);
                        ops.push(OpCode::UpdateLookupIndex {
                            state_id: self.state_id,
                            index_name,
                            lookup_value: lookup_reg,
                            primary_key: key_reg,
                        });
                    }

                    // Also update PDA reverse lookup table if there's a resolver configured for this entity
                    // This allows instruction handlers to look up the primary key from PDA addresses
                    // Only do this when the source path is different (e.g., __account_address -> id.round_address)
                    if source_path_opt.is_some() {
                        ops.push(OpCode::UpdatePdaReverseLookup {
                            state_id: self.state_id,
                            lookup_name: "default_pda_lookup".to_string(),
                            pda_address: lookup_reg,
                            primary_key: key_reg,
                        });
                    }
                }
                KeyResolutionStrategy::Lookup { .. } => {
                    if let Some(mapping) = mappings
                        .iter()
                        .find(|m| m.target_path == lookup_index.field_name)
                    {
                        let value_ops = self.compile_mapping_source(&mapping.source, lookup_reg);
                        if value_ops.is_empty() {
                            continue;
                        }
                        ops.extend(value_ops);
                        if let Some(transform) = &mapping.transform {
                            ops.push(OpCode::Transform {
                                source: lookup_reg,
                                dest: lookup_reg,
                                transformation: transform.clone(),
                            });
                        }
                        ops.push(OpCode::UpdateLookupIndex {
                            state_id: self.state_id,
                            index_name: format!("{}_lookup_index", source_field),
                            lookup_value: lookup_reg,
                            primary_key: key_reg,
                        });
                    }
                }
                KeyResolutionStrategy::Computed { .. }
                | KeyResolutionStrategy::TemporalLookup { .. } => {
                    // Computed and TemporalLookup handlers don't populate lookup indexes
                }
            }
            // Every declared index follows its field's write guard, regardless
            // of how this handler resolved the row.
            let index_name = format!("{}_lookup_index", source_field);
            if self.declares_index(&index_name) {
                if let Some(guard) = index_guard(mappings, &lookup_index.field_name) {
                    let count = ops.len() - start;
                    ops.splice(
                        start..start,
                        [
                            OpCode::EvaluateEventExpr {
                                expr: guard,
                                transform: None,
                                dest: GUARD_REG,
                            },
                            OpCode::SkipUnless {
                                predicate: GUARD_REG,
                                count,
                            },
                        ],
                    );
                }
            }
        }

        ops
    }

    fn get_event_type(&self, source: &SourceSpec) -> String {
        match source {
            SourceSpec::Source { type_name, .. } => type_name.clone(),
        }
    }

    fn compile_instruction_hook_actions(&self, actions: &[HookAction]) -> Vec<OpCode> {
        let mut ops = Vec::new();
        let state_reg = 2;

        for action in actions {
            match action {
                HookAction::SetField {
                    target_field,
                    source,
                    condition,
                } => {
                    // Check if there's a condition - evaluation handled in VM
                    let _ = condition;

                    let temp_reg = 11; // Use register 11 for hook values

                    // Load the source value
                    let load_ops = self.compile_mapping_source(source, temp_reg);
                    ops.extend(load_ops);

                    // Apply transformation if specified in source
                    if let MappingSource::FromSource {
                        transform: Some(transform_type),
                        ..
                    } = source
                    {
                        ops.push(OpCode::Transform {
                            source: temp_reg,
                            dest: temp_reg,
                            transformation: transform_type.clone(),
                        });
                    }

                    // Conditionally set the field based on parsed condition
                    if let Some(cond_expr) = condition {
                        if let Some(parsed) = &cond_expr.parsed {
                            // Generate condition check opcodes
                            let cond_check_ops = self.compile_condition_check(
                                parsed,
                                temp_reg,
                                state_reg,
                                target_field,
                            );
                            ops.extend(cond_check_ops);
                        } else {
                            // No parsed condition, set unconditionally
                            ops.push(OpCode::SetField {
                                object: state_reg,
                                path: target_field.clone(),
                                value: temp_reg,
                            });
                        }
                    } else {
                        // No condition, set unconditionally
                        ops.push(OpCode::SetField {
                            object: state_reg,
                            path: target_field.clone(),
                            value: temp_reg,
                        });
                    }
                }
                HookAction::IncrementField {
                    target_field,
                    increment_by,
                    condition,
                } => {
                    if let Some(cond_expr) = condition {
                        if let Some(parsed) = &cond_expr.parsed {
                            // For increment with condition, we need to:
                            // 1. Load the condition field
                            // 2. Check the condition
                            // 3. Conditionally increment
                            let cond_check_ops = self.compile_conditional_increment(
                                parsed,
                                state_reg,
                                target_field,
                                *increment_by,
                            );
                            ops.extend(cond_check_ops);
                        } else {
                            // No parsed condition, increment unconditionally
                            ops.push(OpCode::SetFieldIncrement {
                                object: state_reg,
                                path: target_field.clone(),
                            });
                        }
                    } else {
                        // No condition, increment unconditionally
                        ops.push(OpCode::SetFieldIncrement {
                            object: state_reg,
                            path: target_field.clone(),
                        });
                    }
                }
                HookAction::RegisterPdaMapping {
                    pda_field,
                    seed_field,
                    lookup_name,
                    sibling,
                    seed_expr,
                    guard,
                } => {
                    let pda_reg = 11;
                    let seed_reg = 12;
                    let mut registration = Vec::new();
                    let pda_from_sibling = sibling
                        .as_ref()
                        .is_some_and(|sibling| sibling.pda_from_sibling);
                    let seed_from_sibling = sibling
                        .as_ref()
                        .is_some_and(|sibling| sibling.seed_from_sibling);

                    // Fields read from this instruction are loaded now; a
                    // sibling's come from its capture.
                    registration.push(if pda_from_sibling {
                        OpCode::LoadConstant {
                            value: Value::Null,
                            dest: pda_reg,
                        }
                    } else {
                        OpCode::LoadEventField {
                            path: pda_field.clone(),
                            dest: pda_reg,
                            default: None,
                        }
                    });
                    registration.push(match (seed_from_sibling, seed_expr) {
                        (true, _) => OpCode::LoadConstant {
                            value: Value::Null,
                            dest: seed_reg,
                        },
                        (false, Some(expr)) => OpCode::EvaluateEventExpr {
                            expr: expr.clone(),
                            transform: None,
                            dest: seed_reg,
                        },
                        (false, None) => OpCode::LoadEventField {
                            path: seed_field.clone(),
                            dest: seed_reg,
                            default: None,
                        },
                    });
                    registration.push(match sibling {
                        None => OpCode::UpdatePdaReverseLookup {
                            state_id: self.state_id,
                            lookup_name: lookup_name.clone(),
                            pda_address: pda_reg,
                            primary_key: seed_reg,
                        },
                        Some(sibling) => OpCode::RegisterPdaFromSibling {
                            state_id: self.state_id,
                            lookup_name: lookup_name.clone(),
                            sibling: sibling.clone(),
                            pda_field: pda_field.clone(),
                            seed_field: seed_field.clone(),
                            pda_address: pda_reg,
                            primary_key: seed_reg,
                        },
                    });

                    if let Some(guard) = guard {
                        ops.push(OpCode::EvaluateEventExpr {
                            expr: guard.clone(),
                            transform: None,
                            dest: GUARD_REG,
                        });
                        ops.push(OpCode::SkipUnless {
                            predicate: GUARD_REG,
                            count: registration.len(),
                        });
                    }
                    ops.extend(registration);
                }
            }
        }

        ops
    }

    fn compile_condition_check(
        &self,
        condition: &ParsedCondition,
        value_reg: Register,
        state_reg: Register,
        target_field: &str,
    ) -> Vec<OpCode> {
        match condition {
            ParsedCondition::Comparison {
                field,
                op,
                value: cond_value,
            } => {
                // Generate ConditionalSetField opcode
                vec![OpCode::ConditionalSetField {
                    object: state_reg,
                    path: target_field.to_string(),
                    value: value_reg,
                    condition_field: field.clone(),
                    condition_op: op.clone(),
                    condition_value: cond_value.clone(),
                }]
            }
            ParsedCondition::Logical { .. } => {
                // Logical conditions not yet supported, fall back to unconditional
                tracing::warn!("Logical conditions not yet supported in instruction hooks");
                vec![OpCode::SetField {
                    object: state_reg,
                    path: target_field.to_string(),
                    value: value_reg,
                }]
            }
            ParsedCondition::Guard { expr } => vec![
                OpCode::EvaluateEventExpr {
                    expr: expr.clone(),
                    transform: None,
                    dest: GUARD_REG,
                },
                OpCode::SkipUnless {
                    predicate: GUARD_REG,
                    count: 1,
                },
                OpCode::SetField {
                    object: state_reg,
                    path: target_field.to_string(),
                    value: value_reg,
                },
            ],
        }
    }

    fn compile_conditional_increment(
        &self,
        condition: &ParsedCondition,
        state_reg: Register,
        target_field: &str,
        _increment_by: i64,
    ) -> Vec<OpCode> {
        match condition {
            ParsedCondition::Comparison {
                field,
                op,
                value: cond_value,
            } => {
                vec![OpCode::ConditionalIncrement {
                    object: state_reg,
                    path: target_field.to_string(),
                    condition_field: field.clone(),
                    condition_op: op.clone(),
                    condition_value: cond_value.clone(),
                }]
            }
            ParsedCondition::Logical { .. } => {
                tracing::warn!("Logical conditions not yet supported in instruction hooks");
                vec![OpCode::SetFieldIncrement {
                    object: state_reg,
                    path: target_field.to_string(),
                }]
            }
            ParsedCondition::Guard { expr } => vec![
                OpCode::EvaluateEventExpr {
                    expr: expr.clone(),
                    transform: None,
                    dest: GUARD_REG,
                },
                OpCode::SkipUnless {
                    predicate: GUARD_REG,
                    count: 1,
                },
                OpCode::SetFieldIncrement {
                    object: state_reg,
                    path: target_field.to_string(),
                },
            ],
        }
    }
}

#[cfg(test)]
mod tests {
    use super::MultiEntityBytecode;
    use crate::ast::{
        FieldPath, HookAction, IdentitySpec, InstructionHook, KeyResolutionStrategy,
        LookupIndexSpec, MappingSource, PopulationStrategy, SerializableFieldMapping,
        SerializableHandlerSpec, SerializableStreamSpec, SourceSpec, TypedStreamSpec,
    };
    use crate::vm::VmContext;
    use crate::{Mutation, WholeEntity};
    use serde_json::{json, Value};
    use std::collections::BTreeMap;

    fn mapping(
        target_path: &str,
        source_path: &[&str],
        population: PopulationStrategy,
    ) -> SerializableFieldMapping {
        SerializableFieldMapping {
            target_path: target_path.to_string(),
            source: MappingSource::FromSource {
                path: FieldPath::new(source_path),
                default: None,
                transform: None,
            },
            transform: None,
            population,
            condition: None,
            when: None,
            stop: None,
            emit: true,
        }
    }

    fn direct_pda_to_primary_key_spec() -> TypedStreamSpec<Value> {
        TypedStreamSpec::from_serializable(SerializableStreamSpec {
            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
            state_name: "PumpfunToken".to_string(),
            program_id: None,
            idl: None,
            identity: IdentitySpec {
                primary_keys: vec!["id.mint".to_string()],
                lookup_indexes: vec![LookupIndexSpec {
                    field_name: "id.bonding_curve".to_string(),
                    temporal_field: None,
                }],
            },
            handlers: vec![SerializableHandlerSpec {
                source: SourceSpec::Source {
                    program_id: None,
                    discriminator: None,
                    type_name: "pump::BondingCurveState".to_string(),
                    serialization: None,
                    is_account: true,
                },
                key_resolution: KeyResolutionStrategy::Lookup {
                    primary_field: FieldPath::new(&["__account_address"]),
                    index_name: None,
                },
                mappings: vec![
                    mapping(
                        "id.bonding_curve",
                        &["__account_address"],
                        PopulationStrategy::SetOnce,
                    ),
                    mapping(
                        "reserves.virtual_token_reserves",
                        &["virtual_token_reserves"],
                        PopulationStrategy::LastWrite,
                    ),
                ],
                conditions: vec![],
                emit: true,
            }],
            sections: vec![],
            field_mappings: BTreeMap::new(),
            resolver_hooks: vec![],
            instruction_hooks: vec![InstructionHook {
                instruction_type: "pump::BuyIxState".to_string(),
                actions: vec![HookAction::RegisterPdaMapping {
                    pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
                    seed_field: FieldPath::new(&["accounts", "mint"]),
                    lookup_name: "default_pda_lookup".to_string(),
                    sibling: None,
                    seed_expr: None,
                    guard: None,
                }],
                lookup_by: None,
            }],
            resolver_specs: vec![],
            computed_fields: vec![],
            computed_field_specs: vec![],
            content_hash: None,
            views: vec![],
            scope_field: None,
        })
    }

    /// A token whose bonding-curve account is keyed by the `id.bonding_curve`
    /// lookup index, which `LinkIxState` writes, or through the PDA mapping
    /// `BuyIxState` registers to the token's creator (indexed by
    /// `CreateIxState`), and whose round account is keyed by the numeric
    /// `id.round` index `OpenRoundIxState` writes. Both account handlers
    /// record an event, a capture and the update's context timestamp.
    fn timed_bonding_curve_spec() -> TypedStreamSpec<Value> {
        let with_source = |target: &str, source: MappingSource| SerializableFieldMapping {
            source,
            ..mapping(target, &[], PopulationStrategy::LastWrite)
        };
        let timestamped = || {
            vec![
                with_source(
                    "events.reserves",
                    MappingSource::AsEvent {
                        fields: vec![Box::new(MappingSource::FromSource {
                            path: FieldPath::new(&["virtual_token_reserves"]),
                            default: None,
                            transform: None,
                        })],
                    },
                ),
                with_source(
                    "snapshots.curve",
                    MappingSource::AsCapture {
                        field_transforms: BTreeMap::new(),
                    },
                ),
                with_source(
                    "state.seen_at",
                    MappingSource::FromContext {
                        field: "timestamp".to_string(),
                    },
                ),
            ]
        };
        let source = |type_name: &str, is_account: bool| SourceSpec::Source {
            program_id: None,
            discriminator: None,
            type_name: type_name.to_string(),
            serialization: None,
            is_account,
        };
        TypedStreamSpec::from_serializable(SerializableStreamSpec {
            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
            state_name: "PumpfunToken".to_string(),
            program_id: None,
            idl: None,
            identity: IdentitySpec {
                primary_keys: vec!["id.mint".to_string()],
                lookup_indexes: vec![
                    LookupIndexSpec {
                        field_name: "id.bonding_curve".to_string(),
                        temporal_field: None,
                    },
                    LookupIndexSpec {
                        field_name: "id.creator".to_string(),
                        temporal_field: None,
                    },
                    LookupIndexSpec {
                        field_name: "id.round".to_string(),
                        temporal_field: None,
                    },
                ],
            },
            handlers: vec![
                SerializableHandlerSpec {
                    source: source("pump::BondingCurveState", true),
                    key_resolution: KeyResolutionStrategy::Lookup {
                        primary_field: FieldPath::new(&["__account_address"]),
                        index_name: None,
                    },
                    mappings: [
                        vec![mapping(
                            "id.bonding_curve",
                            &["__account_address"],
                            PopulationStrategy::SetOnce,
                        )],
                        timestamped(),
                    ]
                    .concat(),
                    conditions: vec![],
                    emit: true,
                },
                SerializableHandlerSpec {
                    source: source("pump::RoundState", true),
                    key_resolution: KeyResolutionStrategy::Lookup {
                        primary_field: FieldPath::new(&["round"]),
                        index_name: None,
                    },
                    mappings: timestamped(),
                    conditions: vec![],
                    emit: true,
                },
                SerializableHandlerSpec {
                    source: source("pump::OpenRoundIxState", false),
                    key_resolution: KeyResolutionStrategy::Embedded {
                        primary_field: FieldPath::new(&["accounts", "mint"]),
                    },
                    mappings: vec![
                        mapping(
                            "id.mint",
                            &["accounts", "mint"],
                            PopulationStrategy::SetOnce,
                        ),
                        mapping(
                            "id.round",
                            &["accounts", "round"],
                            PopulationStrategy::SetOnce,
                        ),
                    ],
                    conditions: vec![],
                    emit: true,
                },
                SerializableHandlerSpec {
                    source: source("pump::LinkIxState", false),
                    key_resolution: KeyResolutionStrategy::Embedded {
                        primary_field: FieldPath::new(&["accounts", "mint"]),
                    },
                    mappings: vec![
                        mapping(
                            "id.mint",
                            &["accounts", "mint"],
                            PopulationStrategy::SetOnce,
                        ),
                        mapping(
                            "id.bonding_curve",
                            &["accounts", "bonding_curve"],
                            PopulationStrategy::SetOnce,
                        ),
                    ],
                    conditions: vec![],
                    emit: true,
                },
                SerializableHandlerSpec {
                    source: source("pump::CreateIxState", false),
                    key_resolution: KeyResolutionStrategy::Embedded {
                        primary_field: FieldPath::new(&["accounts", "mint"]),
                    },
                    mappings: vec![
                        mapping(
                            "id.mint",
                            &["accounts", "mint"],
                            PopulationStrategy::SetOnce,
                        ),
                        mapping(
                            "id.creator",
                            &["accounts", "user"],
                            PopulationStrategy::SetOnce,
                        ),
                    ],
                    conditions: vec![],
                    emit: true,
                },
            ],
            sections: vec![],
            field_mappings: BTreeMap::new(),
            resolver_hooks: vec![],
            instruction_hooks: vec![InstructionHook {
                instruction_type: "pump::BuyIxState".to_string(),
                actions: vec![HookAction::RegisterPdaMapping {
                    pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
                    seed_field: FieldPath::new(&["accounts", "user"]),
                    lookup_name: "default_pda_lookup".to_string(),
                    sibling: None,
                    seed_expr: None,
                    guard: None,
                }],
                lookup_by: None,
            }],
            resolver_specs: vec![],
            computed_fields: vec![],
            computed_field_specs: vec![],
            content_hash: None,
            views: vec![],
            scope_field: None,
        })
    }

    /// A timed account update that waits for its key replays with its own
    /// time: the event, the capture and the `__timestamp` field all carry the
    /// update's timestamp, not the flush time or the flushing instruction's.
    /// The update waits for its lookup index entry (`LinkIxState`), for its
    /// PDA mapping (`BuyIxState`), for the index entry of its already mapped
    /// PDA's seed (`CreateIxState`), or for a numeric index entry, which is
    /// no PDA (`OpenRoundIxState`).
    #[test]
    fn a_queued_account_update_replays_with_its_own_timestamp() {
        use crate::vm::UpdateContext;

        let accounts = json!({
            "bonding_curve": "bonding_curve_1",
            "mint": "mint_1",
            "user": "user_1",
            "round": 7,
        });
        let curve = (
            "pump::BondingCurveState",
            json!({ "__account_address": "bonding_curve_1", "virtual_token_reserves": 42 }),
        );
        let round = (
            "pump::RoundState",
            json!({ "__account_address": "round_1", "round": 7, "virtual_token_reserves": 42 }),
        );
        let instruction = |vm: &mut VmContext, bytecode: &MultiEntityBytecode, name: &str| {
            let mut context =
                UpdateContext::new_instruction(20, "instruction-signature".to_string(), 0);
            context.timestamp = Some(2_000);
            vm.process_event(
                bytecode,
                json!({ "accounts": accounts }),
                name,
                Some(&context),
                None,
            )
            .unwrap();
        };
        // (instructions before the account update, the account update,
        // instructions after it)
        let scenarios: [(&[&str], _, &[&str]); 4] = [
            (&[], &curve, &["pump::LinkIxState"]),
            (&[], &curve, &["pump::CreateIxState", "pump::BuyIxState"]),
            (&["pump::BuyIxState"], &curve, &["pump::CreateIxState"]),
            (&[], &round, &["pump::OpenRoundIxState"]),
        ];
        for (before, (account_type, account), after) in scenarios {
            let bytecode = MultiEntityBytecode::from_single(
                "PumpfunToken".to_string(),
                timed_bonding_curve_spec(),
                0,
            );
            let mut vm = VmContext::new();
            for name in before {
                instruction(&mut vm, &bytecode, name);
            }
            let mut account_context =
                UpdateContext::new_account(10, "curve-signature".to_string(), 1);
            account_context.timestamp = Some(1_000);

            let queued = vm
                .process_event(
                    &bytecode,
                    account.clone(),
                    account_type,
                    Some(&account_context),
                    None,
                )
                .unwrap();
            assert!(queued.is_empty(), "{after:?}: {queued:?}");

            // Flush after a wall-clock second has passed, under instructions
            // with a later timestamp of their own.
            std::thread::sleep(std::time::Duration::from_millis(1_100));
            for name in after {
                instruction(&mut vm, &bytecode, name);
            }

            let token = vm
                .get_entity_state(0, &json!("mint_1"))
                .unwrap_or_else(|| panic!("{after:?}: no entity"));
            for path in [
                "/events/reserves/timestamp",
                "/snapshots/curve/timestamp",
                "/state/seen_at",
            ] {
                assert_eq!(
                    token.pointer(path),
                    Some(&json!(1_000)),
                    "{before:?} {after:?} {path}: {token:#}"
                );
            }
            assert_eq!(
                token.pointer("/events/reserves/data/virtual_token_reserves"),
                Some(&json!(42))
            );
            assert_eq!(token.pointer("/events/reserves/slot"), Some(&json!(10)));
            assert_eq!(
                token.pointer("/snapshots/curve/signature"),
                Some(&json!("curve-signature"))
            );
        }
    }

    fn pda_to_intermediate_lookup_spec() -> TypedStreamSpec<Value> {
        TypedStreamSpec::from_serializable(SerializableStreamSpec {
            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
            state_name: "OreRound".to_string(),
            program_id: None,
            idl: None,
            identity: IdentitySpec {
                primary_keys: vec!["id.round_id".to_string()],
                lookup_indexes: vec![LookupIndexSpec {
                    field_name: "id.round_address".to_string(),
                    temporal_field: None,
                }],
            },
            handlers: vec![SerializableHandlerSpec {
                source: SourceSpec::Source {
                    program_id: None,
                    discriminator: None,
                    type_name: "entropy::VarState".to_string(),
                    serialization: None,
                    is_account: true,
                },
                key_resolution: KeyResolutionStrategy::Lookup {
                    primary_field: FieldPath::new(&["__account_address"]),
                    index_name: None,
                },
                mappings: vec![mapping(
                    "state.expires_at",
                    &["end_at"],
                    PopulationStrategy::LastWrite,
                )],
                conditions: vec![],
                emit: true,
            }],
            sections: vec![],
            field_mappings: BTreeMap::new(),
            resolver_hooks: vec![],
            instruction_hooks: vec![InstructionHook {
                instruction_type: "ore::DeployIxState".to_string(),
                actions: vec![HookAction::RegisterPdaMapping {
                    pda_field: FieldPath::new(&["accounts", "entropyVar"]),
                    seed_field: FieldPath::new(&["accounts", "round"]),
                    lookup_name: "default_pda_lookup".to_string(),
                    sibling: None,
                    seed_expr: None,
                    guard: None,
                }],
                lookup_by: None,
            }],
            resolver_specs: vec![],
            computed_fields: vec![],
            computed_field_specs: vec![],
            content_hash: None,
            views: vec![],
            scope_field: None,
        })
    }

    fn embedded_account_key_spec() -> TypedStreamSpec<Value> {
        TypedStreamSpec::from_serializable(SerializableStreamSpec {
            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
            state_name: "OreTreasury".to_string(),
            program_id: None,
            idl: None,
            identity: IdentitySpec {
                primary_keys: vec!["id.address".to_string()],
                lookup_indexes: vec![],
            },
            handlers: vec![SerializableHandlerSpec {
                source: SourceSpec::Source {
                    program_id: None,
                    discriminator: None,
                    type_name: "ore::TreasuryState".to_string(),
                    serialization: None,
                    is_account: true,
                },
                key_resolution: KeyResolutionStrategy::Embedded {
                    primary_field: FieldPath::new(&["__account_address"]),
                },
                mappings: vec![mapping(
                    "id.address",
                    &["__account_address"],
                    PopulationStrategy::SetOnce,
                )],
                conditions: vec![],
                emit: true,
            }],
            sections: vec![],
            field_mappings: BTreeMap::new(),
            resolver_hooks: vec![],
            instruction_hooks: vec![],
            resolver_specs: vec![],
            computed_fields: vec![],
            computed_field_specs: vec![],
            content_hash: None,
            views: vec![],
            scope_field: None,
        })
    }

    /// A pool whose account carries a price (replaced), a fill (appended) and
    /// a secret that is mapped but never emitted, plus a config account whose
    /// handler emits nothing and sets a fee.
    fn whole_entity_pool_spec() -> TypedStreamSpec<Value> {
        let mut hidden = mapping("state.secret", &["secret"], PopulationStrategy::LastWrite);
        hidden.emit = false;
        let handler = |type_name: &str, mappings, emit| SerializableHandlerSpec {
            source: SourceSpec::Source {
                program_id: None,
                discriminator: None,
                type_name: type_name.to_string(),
                serialization: None,
                is_account: true,
            },
            key_resolution: KeyResolutionStrategy::Embedded {
                primary_field: FieldPath::new(&["__account_address"]),
            },
            mappings,
            conditions: vec![],
            emit,
        };
        TypedStreamSpec::from_serializable(SerializableStreamSpec {
            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
            state_name: "Pool".to_string(),
            program_id: None,
            idl: None,
            identity: IdentitySpec {
                primary_keys: vec!["id.address".to_string()],
                lookup_indexes: vec![],
            },
            handlers: vec![
                handler(
                    "amm::PoolState",
                    vec![
                        mapping(
                            "id.address",
                            &["__account_address"],
                            PopulationStrategy::SetOnce,
                        ),
                        mapping("state.price", &["price"], PopulationStrategy::LastWrite),
                        mapping("state.fills", &["fill"], PopulationStrategy::Append),
                        hidden,
                    ],
                    true,
                ),
                handler(
                    "amm::PoolConfigState",
                    vec![mapping(
                        "config.fee",
                        &["fee"],
                        PopulationStrategy::LastWrite,
                    )],
                    false,
                ),
            ],
            sections: vec![],
            field_mappings: BTreeMap::new(),
            resolver_hooks: vec![],
            instruction_hooks: vec![],
            resolver_specs: vec![],
            computed_fields: vec![],
            computed_field_specs: vec![],
            content_hash: None,
            views: vec![],
            scope_field: None,
        })
    }

    fn pool_bytecode() -> MultiEntityBytecode {
        MultiEntityBytecode::from_single("Pool".to_string(), whole_entity_pool_spec(), 0)
    }

    /// Event (`#[event]`), capture (`#[snapshot]`) and context (`__timestamp`)
    /// mappings compiled from a spec are stamped with the update's timestamp,
    /// so replaying an update reproduces them.
    #[test]
    fn compiled_event_capture_and_context_mappings_carry_the_update_timestamp() {
        let with_source = |target: &str, source: MappingSource| SerializableFieldMapping {
            source,
            ..mapping(target, &[], PopulationStrategy::LastWrite)
        };
        let spec: TypedStreamSpec<Value> =
            TypedStreamSpec::from_serializable(SerializableStreamSpec {
                ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
                state_name: "Pool".to_string(),
                program_id: None,
                idl: None,
                identity: IdentitySpec {
                    primary_keys: vec!["id.address".to_string()],
                    lookup_indexes: vec![],
                },
                handlers: vec![SerializableHandlerSpec {
                    source: SourceSpec::Source {
                        program_id: None,
                        discriminator: None,
                        type_name: "amm::PoolState".to_string(),
                        serialization: None,
                        is_account: true,
                    },
                    key_resolution: KeyResolutionStrategy::Embedded {
                        primary_field: FieldPath::new(&["__account_address"]),
                    },
                    mappings: vec![
                        mapping(
                            "id.address",
                            &["__account_address"],
                            PopulationStrategy::SetOnce,
                        ),
                        with_source(
                            "events.priced",
                            MappingSource::AsEvent {
                                fields: vec![Box::new(MappingSource::FromSource {
                                    path: FieldPath::new(&["price"]),
                                    default: None,
                                    transform: None,
                                })],
                            },
                        ),
                        with_source(
                            "snapshots.pool",
                            MappingSource::AsCapture {
                                field_transforms: BTreeMap::new(),
                            },
                        ),
                        with_source(
                            "state.updated_at",
                            MappingSource::FromContext {
                                field: "timestamp".to_string(),
                            },
                        ),
                    ],
                    conditions: vec![],
                    emit: true,
                }],
                sections: vec![],
                field_mappings: BTreeMap::new(),
                resolver_hooks: vec![],
                instruction_hooks: vec![],
                resolver_specs: vec![],
                computed_fields: vec![],
                computed_field_specs: vec![],
                content_hash: None,
                views: vec![],
                scope_field: None,
            });
        let bytecode = MultiEntityBytecode::from_single("Pool".to_string(), spec, 0);
        let context = crate::vm::UpdateContext::with_timestamp(7, "sig-7".to_string(), 1_234_567);

        let mut vm = VmContext::new();
        vm.process_event(
            &bytecode,
            pool_event(100, 1),
            "amm::PoolState",
            Some(&context),
            None,
        )
        .unwrap();

        let pool = vm.get_entity_state(0, &json!("pool_1")).unwrap();
        for path in [
            "/events/priced/timestamp",
            "/snapshots/pool/timestamp",
            "/state/updated_at",
        ] {
            assert_eq!(
                pool.pointer(path),
                Some(&json!(1_234_567)),
                "{path}: {pool:#}"
            );
        }
        assert_eq!(pool.pointer("/events/priced/data/price"), Some(&json!(100)));
        assert_eq!(pool.pointer("/snapshots/pool/slot"), Some(&json!(7)));
    }

    fn pool_event(price: u64, fill: u64) -> Value {
        pool_event_for("pool_1", price, fill)
    }

    fn pool_event_for(address: &str, price: u64, fill: u64) -> Value {
        json!({"__account_address": address, "price": price, "fill": fill, "secret": "s"})
    }

    /// A VM that answers whole-entity requests, and so marks creations.
    fn linked_vm(
        state_config: crate::vm::StateTableConfig,
    ) -> (VmContext, crate::vm::WholeEntityRequests) {
        let requests = crate::vm::WholeEntityRequests::new(16);
        let mut vm = VmContext::new_with_config(state_config);
        vm.set_whole_entity_requests(requests.clone());
        (vm, requests)
    }

    fn process(vm: &mut VmContext, bytecode: &MultiEntityBytecode, event: Value) -> Vec<Mutation> {
        vm.process_event(bytecode, event, "amm::PoolState", None, None)
            .unwrap()
    }

    #[test]
    fn source_deletion_cancels_resync_and_recreates_a_fresh_vm_row() {
        let bytecode = pool_bytecode();
        let (mut vm, requests) = linked_vm(Default::default());
        let first = process(&mut vm, &bytecode, pool_event(1, 10));
        let key = &first[0].key;
        let target = crate::vm::ResolverTarget {
            state_id: 0,
            entity_name: "Pool".into(),
            primary_key: key.clone(),
            extracts: vec![],
        };
        vm.enqueue_resolver_request(
            "ignored".into(),
            crate::ast::ResolverType::Token,
            json!("mint"),
            target.clone(),
        );
        let old_request = vm.take_resolver_requests().pop().unwrap();
        vm.get_state_table_mut(0).unwrap().deferred_when_ops.insert(
            ("sig".into(), "instruction".into()),
            vec![crate::vm::DeferredWhenOperation {
                entity_name: "Pool".into(),
                primary_key: key.clone(),
                field_path: "old".into(),
                field_value: json!(true),
                when_instruction: "instruction".into(),
                signature: "sig".into(),
                slot: 1,
                deferred_at: 1,
                emit: true,
            }],
        );
        requests.request("Pool", key);
        let deletion = vm.delete_entity(&bytecode, "Pool", key).unwrap();
        assert!(deletion.is_delete());
        assert!(
            serde_json::from_value::<Mutation>(serde_json::to_value(&deletion).unwrap())
                .unwrap()
                .is_delete()
        );
        assert!(!requests.is_requested("Pool", key));
        assert!(vm.get_entity_state(0, key).is_none());
        assert!(vm.snapshot_state_table(0).is_empty());
        assert!(vm
            .get_state_table_mut(0)
            .unwrap()
            .deferred_when_ops
            .is_empty());
        assert!(vm.delete_entity(&bytecode, "Unknown", key).is_none());
        let mut next = process(&mut vm, &bytecode, pool_event(9, 90));
        vm.enqueue_resolver_request(
            "ignored".into(),
            crate::ast::ResolverType::Token,
            json!("mint"),
            target,
        );
        let new_request = vm.take_resolver_requests().pop().unwrap();
        assert_ne!(old_request.cache_key, new_request.cache_key);
        assert!(vm
            .apply_resolver_result(&bytecode, &old_request.cache_key, json!({"stale":true}))
            .unwrap()
            .is_empty());
        assert_eq!(next[0].take_whole_entity_mark(), Some(WholeEntity::Created));
        assert_eq!(next[0].patch["state"]["fills"], json!([90]));
    }

    #[test]
    fn deletion_preserves_unrelated_resolver_sharing_and_isolates_recreation() {
        let bytecode = pool_bytecode();
        let (mut vm, _) = linked_vm(Default::default());
        let key = process(&mut vm, &bytecode, pool_event(1, 10))[0]
            .key
            .clone();
        let late_joiner = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20))[0]
            .key
            .clone();
        let target = |primary_key| crate::vm::ResolverTarget {
            state_id: 0,
            entity_name: "Pool".into(),
            primary_key,
            extracts: vec![crate::ast::ResolverExtractSpec {
                target_path: "resolved".into(),
                source_path: Some("value".into()),
                transform: None,
            }],
        };
        let enqueue = |vm: &mut VmContext, key| {
            vm.enqueue_resolver_request(
                "ignored".into(),
                crate::ast::ResolverType::Token,
                json!("shared-mint"),
                target(key),
            )
        };
        enqueue(&mut vm, key.clone());
        enqueue(&mut vm, json!("unrelated"));
        let old = vm.take_resolver_requests();
        assert_eq!(old.len(), 1);
        vm.delete_entity(&bytecode, "Pool", &key).unwrap();
        enqueue(&mut vm, json!("unrelated"));
        enqueue(&mut vm, json!("another-unrelated"));
        assert!(
            vm.take_resolver_requests().is_empty(),
            "unaffected targets reuse the old pending call"
        );
        enqueue(&mut vm, key.clone());
        let fresh = vm.take_resolver_requests();
        assert_eq!(fresh.len(), 1);
        assert_ne!(fresh[0].cache_key, old[0].cache_key);
        // This entity never joined the old shared call. Deleting and
        // recreating it must still keep it off that pre-deletion result.
        vm.delete_entity(&bytecode, "Pool", &late_joiner).unwrap();
        process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
        enqueue(&mut vm, late_joiner.clone());
        let late_joiner_fresh = vm.take_resolver_requests();
        assert_eq!(late_joiner_fresh.len(), 1);
        assert_ne!(late_joiner_fresh[0].cache_key, old[0].cache_key);
        enqueue(&mut vm, json!("unrelated"));
        enqueue(&mut vm, json!("another-unrelated"));
        enqueue(&mut vm, key.clone());
        assert!(
            vm.take_resolver_requests().is_empty(),
            "unaffected targets reuse pending calls"
        );
        let old_mutations = vm
            .apply_resolver_result(&bytecode, &old[0].cache_key, json!({"value":"old"}))
            .unwrap();
        assert_eq!(old_mutations.len(), 2);
        assert!(old_mutations.iter().all(|mutation| mutation.key != key));
        assert!(vm.get_entity_state(0, &key).is_none());
        let fresh_mutations = vm
            .apply_resolver_result(&bytecode, &fresh[0].cache_key, json!({"value":"fresh"}))
            .unwrap();
        assert_eq!(fresh_mutations.len(), 1);
        assert_eq!(fresh_mutations[0].key, key);
        let late_joiner_mutations = vm
            .apply_resolver_result(
                &bytecode,
                &late_joiner_fresh[0].cache_key,
                json!({"value":"fresh-late-joiner"}),
            )
            .unwrap();
        assert_eq!(late_joiner_mutations.len(), 1);
        assert_eq!(late_joiner_mutations[0].key, late_joiner);
    }

    #[test]
    fn a_requested_entity_is_emitted_whole_once() {
        let bytecode = pool_bytecode();
        let (mut vm, requests) = linked_vm(Default::default());
        let mut update = |price, fill| process(&mut vm, &bytecode, pool_event(price, fill));

        let first = update(1, 10);
        assert_eq!(first[0].whole_entity_mark(), Some(WholeEntity::Created));
        let key = first[0].key.clone();
        let second = update(2, 20);
        assert_eq!(second.len(), 1);
        assert!(!second[0].is_whole_entity());
        assert_eq!(
            second[0].patch,
            json!({"state": {"price": 2, "fills": [20]}})
        );
        assert_eq!(second[0].append, vec!["state.fills".to_string()]);

        // A consumer that dropped the entity asks for all of it.
        requests.request("Pool", &key);
        assert!(requests.is_requested("Pool", &key));
        let mut third = update(3, 30);
        assert_eq!(third.len(), 2);
        assert!(
            !third[0].is_whole_entity(),
            "the change itself is untouched"
        );
        assert_eq!(
            third[0].patch,
            json!({"state": {"price": 3, "fills": [30]}})
        );
        assert_eq!(third[1].key, key);
        assert_eq!(third[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
        assert_eq!(
            third[1].patch,
            json!({
                "id": {"address": "pool_1"},
                "state": {"price": 3, "fills": [10, 20, 30]},
            }),
            "the whole entity, without its non-emitted field"
        );
        assert!(
            third[1].append.is_empty(),
            "its arrays are whole, not appended to"
        );
        assert!(requests.is_empty(), "a request is answered once");

        let fourth = update(4, 40);
        assert_eq!(fourth.len(), 1);
        assert!(!fourth[0].is_whole_entity());
        assert_eq!(
            fourth[0].patch,
            json!({"state": {"price": 4, "fills": [40]}})
        );
    }

    /// A request does not wait for its entity to change: the VM's next call
    /// ends with the whole entity, after whatever that call emitted.
    #[test]
    fn a_request_is_answered_by_the_next_call_whatever_it_changes() {
        let bytecode = pool_bytecode();
        let (mut vm, requests) = linked_vm(Default::default());
        process(&mut vm, &bytecode, pool_event_for("pool_1", 1, 10));
        process(&mut vm, &bytecode, pool_event_for("pool_2", 1, 10));

        requests.request("Pool", &json!("pool_1"));
        let mut next = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20));
        assert_eq!(next.len(), 2);
        assert_eq!(next[0].key, json!("pool_2"));
        assert!(!next[0].is_whole_entity());
        assert_eq!(next[1].key, json!("pool_1"));
        assert_eq!(next[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
        assert_eq!(
            next[1].patch,
            json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
        );
        assert!(requests.is_empty());

        // Even a call that changes nothing answers.
        requests.request("Pool", &json!("pool_2"));
        let answered = vm
            .apply_resolver_result(&bytecode, "no-such-request", json!({}))
            .unwrap();
        assert_eq!(answered.len(), 1);
        assert_eq!(answered[0].key, json!("pool_2"));
        assert_eq!(answered[0].whole_entity_mark(), Some(WholeEntity::Resent));

        let quiet = process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
        assert_eq!(quiet.len(), 1, "a request is answered once");
    }

    /// The VM has nothing to send for an entity it does not hold. It dropped
    /// the entity too, so the key's next mutation creates it again.
    #[test]
    fn a_request_for_an_entity_the_vm_does_not_hold_is_dropped() {
        let bytecode = pool_bytecode();
        let (mut vm, requests) = linked_vm(Default::default());
        requests.request("Pool", &json!("other_pool"));
        let mutations = process(&mut vm, &bytecode, pool_event(1, 10));
        assert_eq!(mutations.len(), 1);
        assert_eq!(mutations[0].whole_entity_mark(), Some(WholeEntity::Created));
        assert!(requests.is_empty(), "dropped, not kept for later");

        let created = process(&mut vm, &bytecode, pool_event_for("other_pool", 1, 10));
        assert_eq!(created.len(), 1);
        assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
    }

    #[test]
    fn whole_entity_requests_are_bounded() {
        let requests = crate::vm::WholeEntityRequests::new(2);
        for key in ["a", "b", "c"] {
            requests.request("Pool", &json!(key));
        }
        assert_eq!(requests.len(), 2);
        assert!(
            !requests.is_requested("Pool", &json!("a")),
            "the oldest is forgotten"
        );
        assert!(requests.is_requested("Pool", &json!("c")));
    }

    /// A consumer that does not hold a key can tell a new entity from a later
    /// patch for one it dropped: only the first is marked created.
    #[test]
    fn an_entitys_first_mutation_is_marked_as_its_creation() {
        let bytecode = pool_bytecode();
        let (mut vm, _requests) = linked_vm(Default::default());

        let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
        assert_eq!(first.len(), 1, "the creation needs no second mutation");
        assert_eq!(
            first[0].take_whole_entity_mark(),
            Some(WholeEntity::Created)
        );
        assert_eq!(
            first[0].patch,
            json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}}),
            "the whole row, without its non-emitted field"
        );
        assert_eq!(first[0].append, vec!["state.fills".to_string()]);

        let second = process(&mut vm, &bytecode, pool_event(2, 20));
        assert_eq!(second[0].whole_entity_mark(), None);
        assert_eq!(
            second[0].patch,
            json!({"state": {"price": 2, "fills": [20]}})
        );
    }

    /// Without a consumer asking for whole entities the VM's output is
    /// unchanged: nothing is marked, and a creation is its changed fields.
    #[test]
    fn a_vm_without_whole_entity_requests_marks_no_creation() {
        let bytecode = pool_bytecode();
        let mut vm = VmContext::new();
        let first = process(&mut vm, &bytecode, pool_event(1, 10));
        assert!(!first[0].is_whole_entity());
        assert!(first[0].patch.get(crate::WHOLE_ENTITY_MARKER).is_none());
    }

    /// A handler that emits nothing can start the row. The first mutation
    /// after it is still the creation, and carries what that handler set,
    /// which its own change does not.
    #[test]
    fn a_row_started_without_a_mutation_is_created_by_its_first_one() {
        let bytecode = pool_bytecode();
        let (mut vm, _requests) = linked_vm(Default::default());
        let config = vm
            .process_event(
                &bytecode,
                json!({"__account_address": "pool_1", "fee": 30}),
                "amm::PoolConfigState",
                None,
                None,
            )
            .unwrap();
        assert!(config.is_empty(), "the config handler emits nothing");

        let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
        assert_eq!(
            first[0].take_whole_entity_mark(),
            Some(WholeEntity::Created)
        );
        assert_eq!(
            first[0].patch,
            json!({
                "id": {"address": "pool_1"},
                "config": {"fee": 30},
                "state": {"price": 1, "fills": [10]},
            })
        );
        let second = process(&mut vm, &bytecode, pool_event(2, 20));
        assert!(!second[0].is_whole_entity());
    }

    /// An entity the VM's bounded table dropped starts again from its next
    /// event, and that event's mutation creates it again.
    #[test]
    fn an_entity_the_vm_dropped_is_created_again() {
        let bytecode = pool_bytecode();
        let (mut vm, _requests) = linked_vm(crate::vm::StateTableConfig {
            max_entries: 2,
            ..Default::default()
        });
        for pool in ["a", "b", "c"] {
            let created = process(&mut vm, &bytecode, pool_event_for(pool, 1, 1));
            assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
        }
        assert!(vm.get_entity_state(0, &json!("a")).is_none(), "dropped");

        let mut again = process(&mut vm, &bytecode, pool_event_for("a", 2, 2));
        assert_eq!(
            again[0].take_whole_entity_mark(),
            Some(WholeEntity::Created)
        );
        assert_eq!(
            again[0].patch,
            json!({"id": {"address": "a"}, "state": {"price": 2, "fills": [2]}})
        );
        let changed = process(&mut vm, &bytecode, pool_event_for("c", 2, 2));
        assert!(!changed[0].is_whole_entity(), "c was never dropped");
    }

    /// A row a non-emitting handler started and a snapshot then took before
    /// its first mutation comes back as emitted: its first update after the
    /// restore is a plain patch. A consumer that never held the key asks for
    /// it, and the VM's next call sends it whole, with what that handler set.
    #[test]
    fn a_restored_row_never_emitted_is_sent_whole_on_request() {
        let bytecode = pool_bytecode();
        let (mut vm, _requests) = linked_vm(Default::default());
        let config = vm
            .process_event(
                &bytecode,
                json!({"__account_address": "pool_1", "fee": 30}),
                "amm::PoolConfigState",
                None,
                None,
            )
            .unwrap();
        assert!(config.is_empty());
        let snapshot = vm.dump();

        let (mut restored, requests) = linked_vm(Default::default());
        restored.hydrate(snapshot);
        let first = process(&mut restored, &bytecode, pool_event(1, 10));
        assert!(!first[0].is_whole_entity(), "counted as emitted already");
        assert_eq!(
            first[0].patch,
            json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
        );

        requests.request("Pool", &first[0].key);
        let next = process(&mut restored, &bytecode, pool_event_for("pool_2", 1, 1));
        let resent = next
            .iter()
            .find(|mutation| mutation.key == json!("pool_1"))
            .expect("the requested entity follows the next call");
        assert_eq!(resent.whole_entity_mark(), Some(WholeEntity::Resent));
        let mut resent = resent.clone();
        resent.take_whole_entity_mark();
        assert_eq!(
            resent.patch,
            json!({
                "id": {"address": "pool_1"},
                "config": {"fee": 30},
                "state": {"price": 1, "fills": [10]},
            })
        );
    }

    /// A resolver result for an entity the table does not hold starts the
    /// entity, so its mutation is the creation; one for an entity the table
    /// holds is a patch.
    #[test]
    fn a_resolver_result_for_an_entity_not_held_creates_it() {
        let bytecode = pool_bytecode();
        let (mut vm, _requests) = linked_vm(Default::default());
        process(&mut vm, &bytecode, pool_event(1, 10));
        for key in ["pool_1", "pool_2"] {
            vm.enqueue_resolver_request(
                String::new(),
                crate::ast::ResolverType::Token,
                json!(key),
                crate::vm::ResolverTarget {
                    state_id: 0,
                    entity_name: "Pool".to_string(),
                    primary_key: json!(key),
                    extracts: vec![crate::ast::ResolverExtractSpec {
                        target_path: "meta.name".to_string(),
                        source_path: Some("name".to_string()),
                        transform: None,
                    }],
                },
            );
        }
        let mut applied = Vec::new();
        for request in vm.take_resolver_requests() {
            applied.extend(
                vm.apply_resolver_result(&bytecode, &request.cache_key, json!({"name": "n"}))
                    .unwrap(),
            );
        }
        applied.sort_by_key(|mutation| mutation.key.to_string());

        assert_eq!(applied[0].key, json!("pool_1"));
        assert!(!applied[0].is_whole_entity());
        assert_eq!(applied[0].patch, json!({"meta": {"name": "n"}}));
        assert_eq!(applied[1].key, json!("pool_2"));
        assert_eq!(
            applied[1].take_whole_entity_mark(),
            Some(WholeEntity::Created)
        );
        assert_eq!(applied[1].patch, json!({"meta": {"name": "n"}}));
    }

    /// Entities restored from a snapshot may be held by the consumer already,
    /// so their next mutation is a patch like any other.
    #[test]
    fn a_restored_entity_is_not_created_again() {
        let bytecode = pool_bytecode();
        let (mut vm, _requests) = linked_vm(Default::default());
        process(&mut vm, &bytecode, pool_event(1, 10));
        let snapshot = vm.dump();

        let (mut restored, _requests) = linked_vm(Default::default());
        restored.hydrate(snapshot);
        let next = process(&mut restored, &bytecode, pool_event(2, 20));
        assert!(!next[0].is_whole_entity());
        assert_eq!(next[0].patch, json!({"state": {"price": 2, "fills": [20]}}));
    }

    #[test]
    fn lookup_account_handler_uses_resolved_primary_key_when_hook_seeds_primary_key() {
        let bytecode = MultiEntityBytecode::from_single(
            "PumpfunToken".to_string(),
            direct_pda_to_primary_key_spec(),
            0,
        );
        let mut vm = VmContext::new();

        let mutations = vm
            .process_event(
                &bytecode,
                json!({
                    "__account_address": "bonding_curve_1",
                    "__resolved_primary_key": "mint_1",
                    "virtual_token_reserves": 42,
                }),
                "pump::BondingCurveState",
                None,
                None,
            )
            .unwrap();

        assert_eq!(mutations.len(), 1);
        assert_eq!(mutations[0].key, json!("mint_1"));
        assert_eq!(
            mutations[0].patch["id"]["bonding_curve"],
            json!("bonding_curve_1")
        );
        assert_eq!(
            mutations[0].patch["reserves"]["virtual_token_reserves"],
            json!(42)
        );
    }

    #[test]
    fn lookup_account_handler_keeps_null_key_when_resolver_only_returns_intermediate_lookup() {
        let bytecode = MultiEntityBytecode::from_single(
            "OreRound".to_string(),
            pda_to_intermediate_lookup_spec(),
            0,
        );
        let mut vm = VmContext::new();

        let mutations = vm
            .process_event(
                &bytecode,
                json!({
                    "__account_address": "entropy_var_1",
                    "__resolved_primary_key": "round_address_1",
                    "end_at": 123,
                }),
                "entropy::VarState",
                None,
                None,
            )
            .unwrap();

        assert!(mutations.is_empty());
    }

    #[test]
    fn embedded_account_key_wins_over_an_unrelated_resolved_key() {
        let bytecode = MultiEntityBytecode::from_single(
            "OreTreasury".to_string(),
            embedded_account_key_spec(),
            0,
        );
        let mut vm = VmContext::new();

        let mutations = vm
            .process_event(
                &bytecode,
                json!({
                    "__account_address": "treasury_pda",
                    "__resolved_primary_key": "round_address",
                }),
                "ore::TreasuryState",
                None,
                None,
            )
            .unwrap();

        assert_eq!(mutations.len(), 1);
        assert_eq!(mutations[0].key, json!("treasury_pda"));
        assert_eq!(mutations[0].patch["id"]["address"], json!("treasury_pda"));
    }

    fn instruction_handler(
        type_name: &str,
        key_resolution: KeyResolutionStrategy,
        mappings: Vec<SerializableFieldMapping>,
    ) -> SerializableHandlerSpec {
        SerializableHandlerSpec {
            source: SourceSpec::Source {
                program_id: None,
                discriminator: None,
                type_name: type_name.to_string(),
                serialization: None,
                is_account: false,
            },
            key_resolution,
            mappings,
            conditions: vec![],
            emit: true,
        }
    }

    /// `split_position` updates two positions: the source (`first_position`)
    /// and the child (`second_position`). Mirrors the meteora-damm
    /// `MeteoraPosition` shape: one Count per side plus a derive_from hook on
    /// the source side.
    fn split_position_spec() -> TypedStreamSpec<Value> {
        let embedded = |field: &str| KeyResolutionStrategy::Embedded {
            primary_field: FieldPath::new(&["accounts", field]),
        };
        TypedStreamSpec::from_serializable(SerializableStreamSpec {
            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
            state_name: "Position".to_string(),
            program_id: None,
            idl: None,
            identity: IdentitySpec {
                primary_keys: vec!["id.position_address".to_string()],
                lookup_indexes: vec![],
            },
            handlers: vec![
                instruction_handler(
                    "amm::SplitPositionIxState",
                    embedded("first_position"),
                    vec![
                        mapping(
                            "id.position_address",
                            &["accounts", "first_position"],
                            PopulationStrategy::SetOnce,
                        ),
                        mapping(
                            "activity.split_source_count",
                            &["data"],
                            PopulationStrategy::Count,
                        ),
                    ],
                ),
                instruction_handler(
                    "amm::SplitPositionIxState",
                    embedded("second_position"),
                    vec![
                        mapping(
                            "id.position_address",
                            &["accounts", "second_position"],
                            PopulationStrategy::SetOnce,
                        ),
                        mapping(
                            "activity.split_child_count",
                            &["data"],
                            PopulationStrategy::Count,
                        ),
                    ],
                ),
            ],
            sections: vec![],
            field_mappings: BTreeMap::new(),
            resolver_hooks: vec![],
            instruction_hooks: vec![InstructionHook {
                instruction_type: "amm::SplitPositionIxState".to_string(),
                actions: vec![HookAction::SetField {
                    target_field: "activity.last_split_source_slot".to_string(),
                    source: MappingSource::FromSource {
                        path: FieldPath::new(&["data", "slot"]),
                        default: None,
                        transform: None,
                    },
                    condition: None,
                }],
                lookup_by: Some(FieldPath::new(&["accounts", "first_position"])),
            }],
            resolver_specs: vec![],
            computed_fields: vec![],
            computed_field_specs: vec![],
            content_hash: None,
            views: vec![],
            scope_field: None,
        })
    }

    fn split_position_event(first: &str, second: &str) -> Value {
        json!({
            "accounts": { "first_position": first, "second_position": second },
            "data": { "slot": 77 },
        })
    }

    #[test]
    fn one_instruction_routes_each_key_to_its_own_entity() {
        let bytecode =
            MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
        let mut vm = VmContext::new();

        for _ in 0..2 {
            vm.process_event(
                &bytecode,
                split_position_event("source_pos", "child_pos"),
                "amm::SplitPositionIxState",
                None,
                None,
            )
            .unwrap();
        }
        vm.process_event(
            &bytecode,
            split_position_event("child_pos", "grandchild_pos"),
            "amm::SplitPositionIxState",
            None,
            None,
        )
        .unwrap();

        let state = |key: &str| vm.get_entity_state(0, &json!(key)).unwrap();

        let source = state("source_pos");
        assert_eq!(source["id"]["position_address"], json!("source_pos"));
        assert_eq!(source["activity"]["split_source_count"], json!(2));
        assert_eq!(source["activity"].get("split_child_count"), None);
        assert_eq!(source["activity"]["last_split_source_slot"], json!(77));

        // The child was split into once and split from once.
        let child = state("child_pos");
        assert_eq!(child["id"]["position_address"], json!("child_pos"));
        assert_eq!(child["activity"]["split_child_count"], json!(2));
        assert_eq!(child["activity"]["split_source_count"], json!(1));
        assert_eq!(child["activity"]["last_split_source_slot"], json!(77));

        let grandchild = state("grandchild_pos");
        assert_eq!(grandchild["activity"]["split_child_count"], json!(1));
        assert_eq!(grandchild["activity"].get("split_source_count"), None);
        assert_eq!(grandchild["activity"].get("last_split_source_slot"), None);
    }

    #[test]
    fn one_instruction_emits_one_mutation_per_key() {
        let bytecode =
            MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
        let mut vm = VmContext::new();

        let mutations = vm
            .process_event(
                &bytecode,
                split_position_event("source_pos", "child_pos"),
                "amm::SplitPositionIxState",
                None,
                None,
            )
            .unwrap();

        assert_eq!(mutations.len(), 2);
        assert_eq!(mutations[0].key, json!("source_pos"));
        assert_eq!(
            mutations[0].patch["activity"]["split_source_count"],
            json!(1)
        );
        assert_eq!(
            mutations[0].patch["activity"]["last_split_source_slot"],
            json!(77)
        );
        assert_eq!(
            mutations[0].patch["activity"].get("split_child_count"),
            None
        );
        assert_eq!(mutations[1].key, json!("child_pos"));
        assert_eq!(
            mutations[1].patch["activity"]["split_child_count"],
            json!(1)
        );
        assert_eq!(
            mutations[1].patch["activity"].get("split_source_count"),
            None
        );
    }

    #[test]
    fn handlers_with_identical_keys_still_merge_into_one_segment() {
        let mut spec = split_position_spec();
        spec.instruction_hooks.clear();
        // Re-key the second handler by `first_position`: both now resolve the
        // same key and must share a single state read/write.
        spec.handlers[1].key_resolution = KeyResolutionStrategy::Embedded {
            primary_field: FieldPath::new(&["accounts", "first_position"]),
        };
        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
        let handler = &bytecode.entities["Position"].handlers["amm::SplitPositionIxState"];

        assert!(!handler
            .iter()
            .any(|op| matches!(op, super::OpCode::SegmentBoundary)));
        assert_eq!(
            handler
                .iter()
                .filter(|op| matches!(op, super::OpCode::ReadOrInitState { .. }))
                .count(),
            1
        );
    }

    #[test]
    fn hook_without_a_matching_key_gets_its_own_segment() {
        let mut spec = split_position_spec();
        spec.handlers.truncate(1);
        spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
            primary_field: FieldPath::new(&["accounts", "second_position"]),
        };
        spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
            "id.position_address",
            &["accounts", "second_position"],
            PopulationStrategy::SetOnce,
        ));
        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
        let mut vm = VmContext::new();

        vm.process_event(
            &bytecode,
            split_position_event("source_pos", "child_pos"),
            "amm::SplitPositionIxState",
            None,
            None,
        )
        .unwrap();

        // The hook routes by `first_position`, not by the handler's key.
        let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
        assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
        let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
        assert_eq!(child["activity"].get("last_split_source_slot"), None);
    }

    /// Index positions by owner through a `PositionState` account handler.
    fn add_owner_index(spec: &mut TypedStreamSpec<Value>) {
        spec.identity.lookup_indexes.push(LookupIndexSpec {
            field_name: "id.owner".to_string(),
            temporal_field: None,
        });
        spec.handlers
            .push(crate::ast::TypedHandlerSpec::from_serializable(
                SerializableHandlerSpec {
                    source: SourceSpec::Source {
                        program_id: None,
                        discriminator: None,
                        type_name: "amm::PositionState".to_string(),
                        serialization: None,
                        is_account: true,
                    },
                    key_resolution: KeyResolutionStrategy::Embedded {
                        primary_field: FieldPath::new(&["__account_address"]),
                    },
                    mappings: vec![
                        mapping(
                            "id.position_address",
                            &["__account_address"],
                            PopulationStrategy::SetOnce,
                        ),
                        mapping("id.owner", &["owner"], PopulationStrategy::SetOnce),
                    ],
                    conditions: vec![],
                    emit: true,
                },
            ));
    }

    fn index_owner(
        vm: &mut VmContext,
        bytecode: &MultiEntityBytecode,
        position: &str,
        owner: &str,
    ) {
        vm.process_event(
            bytecode,
            json!({ "__account_address": position, "owner": owner }),
            "amm::PositionState",
            None,
            None,
        )
        .unwrap();
    }

    #[test]
    fn unmatched_hook_next_to_other_keys_resolves_lookup_index_fields() {
        let mut spec = split_position_spec();
        spec.handlers.truncate(1);
        spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
            primary_field: FieldPath::new(&["accounts", "second_position"]),
        };
        spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
            "id.position_address",
            &["accounts", "second_position"],
            PopulationStrategy::SetOnce,
        ));
        add_owner_index(&mut spec);
        // The hook routes by the owner, a lookup-index field.
        spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));

        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
        let mut vm = VmContext::new();
        index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
        vm.process_event(
            &bytecode,
            json!({
                "accounts": {
                    "first_position": "source_pos",
                    "second_position": "child_pos",
                    "owner": "owner_1",
                },
                "data": { "slot": 77 },
            }),
            "amm::SplitPositionIxState",
            None,
            None,
        )
        .unwrap();

        let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
        assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
        assert!(vm.get_entity_state(0, &json!("owner_1")).is_none());
    }

    #[test]
    fn unmatched_hook_replays_after_its_lookup_index_is_populated() {
        let mut spec = split_position_spec();
        spec.handlers.truncate(1);
        spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
            primary_field: FieldPath::new(&["accounts", "second_position"]),
        };
        spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
            "id.position_address",
            &["accounts", "second_position"],
            PopulationStrategy::SetOnce,
        ));
        add_owner_index(&mut spec);
        spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));

        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
        let mut vm = VmContext::new();
        let mutations = vm
            .process_event(
                &bytecode,
                json!({
                    "accounts": { "owner": "owner_1", "second_position": "child_pos" },
                    "data": { "slot": 77 },
                }),
                "amm::SplitPositionIxState",
                None,
                None,
            )
            .unwrap();
        assert_eq!(mutations.len(), 1);
        assert_eq!(mutations[0].key, json!("child_pos"));

        index_owner(&mut vm, &bytecode, "source_pos", "owner_1");

        let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
        assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
        let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
        assert_eq!(child["activity"].get("last_split_source_slot"), None);
    }

    #[test]
    fn a_segment_that_misses_its_key_is_replayed_alone() {
        let mut spec = split_position_spec();
        spec.instruction_hooks.clear();
        // The child segment comes first. The source segment resolves its key
        // through a PDA, registered later, that maps to the position's owner.
        spec.handlers.swap(0, 1);
        spec.handlers[1].key_resolution = KeyResolutionStrategy::Lookup {
            primary_field: FieldPath::new(&["accounts", "source_pda"]),
            index_name: None,
        };
        spec.handlers[1].mappings.remove(0);
        add_owner_index(&mut spec);
        spec.instruction_hooks.push(InstructionHook {
            instruction_type: "amm::RegisterIxState".to_string(),
            actions: vec![HookAction::RegisterPdaMapping {
                pda_field: FieldPath::new(&["accounts", "source_pda"]),
                seed_field: FieldPath::new(&["accounts", "owner"]),
                lookup_name: "default_pda_lookup".to_string(),
                sibling: None,
                seed_expr: None,
                guard: None,
            }],
            lookup_by: None,
        });
        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
        let mut vm = VmContext::new();
        index_owner(&mut vm, &bytecode, "source_pos", "owner_1");

        let mutations = vm
            .process_event(
                &bytecode,
                json!({
                    "accounts": { "source_pda": "pda_1", "second_position": "child_pos" },
                    "data": {},
                }),
                "amm::SplitPositionIxState",
                None,
                None,
            )
            .unwrap();
        // The child landed; the source is not resolvable yet.
        assert_eq!(mutations.len(), 1);
        assert_eq!(mutations[0].key, json!("child_pos"));

        let mutations = vm
            .process_event(
                &bytecode,
                json!({ "accounts": { "source_pda": "pda_1", "owner": "owner_1" } }),
                "amm::RegisterIxState",
                None,
                None,
            )
            .unwrap();
        // Registering the PDA replays only the source segment.
        assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
        assert_eq!(mutations[0].key, json!("source_pos"));
        let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
        assert_eq!(source["activity"]["split_source_count"], json!(1));
        let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
        assert_eq!(child["activity"]["split_child_count"], json!(1));
    }

    /// A hook whose `lookup_by` names a field the instruction does not carry
    /// loads a null key. The entity's key transform must pass that null to
    /// the segment's null-key check, which skips the hook, rather than fail
    /// the event and lose the handler segment's update with it.
    #[test]
    fn a_hook_whose_lookup_field_is_absent_skips_only_its_segment() {
        let spec: TypedStreamSpec<Value> =
            TypedStreamSpec::from_serializable(SerializableStreamSpec {
                ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
                state_name: "Game".to_string(),
                program_id: None,
                idl: None,
                identity: IdentitySpec {
                    primary_keys: vec!["id.game_id".to_string()],
                    lookup_indexes: vec![],
                },
                handlers: vec![instruction_handler(
                    "game::StartGameIxState",
                    KeyResolutionStrategy::Embedded {
                        primary_field: FieldPath::new(&["data", "game_id"]),
                    },
                    vec![
                        SerializableFieldMapping {
                            source: MappingSource::FromSource {
                                path: FieldPath::new(&["data", "game_id"]),
                                default: None,
                                transform: Some(crate::ast::Transformation::HexEncode),
                            },
                            ..mapping("id.game_id", &[], PopulationStrategy::SetOnce)
                        },
                        mapping("stats.fee", &["data", "fee"], PopulationStrategy::SetOnce),
                    ],
                )],
                sections: vec![],
                field_mappings: BTreeMap::new(),
                resolver_hooks: vec![],
                instruction_hooks: vec![InstructionHook {
                    instruction_type: "game::StartGameIxState".to_string(),
                    actions: vec![HookAction::SetField {
                        target_field: "stats.started_slot".to_string(),
                        source: MappingSource::FromSource {
                            path: FieldPath::new(&["data", "slot"]),
                            default: None,
                            transform: None,
                        },
                        condition: None,
                    }],
                    // `game_id` is an argument, not an account.
                    lookup_by: Some(FieldPath::new(&["accounts", "game_id"])),
                }],
                resolver_specs: vec![],
                computed_fields: vec![],
                computed_field_specs: vec![],
                content_hash: None,
                views: vec![],
                scope_field: None,
            });
        let bytecode = MultiEntityBytecode::from_single("Game".to_string(), spec, 0);
        let mut vm = VmContext::new();

        let mutations = vm
            .process_event(
                &bytecode,
                json!({
                    "accounts": { "game_account": "game_pda" },
                    "data": { "game_id": [1, 2, 3, 4], "fee": 5, "slot": 9 },
                }),
                "game::StartGameIxState",
                None,
                None,
            )
            .unwrap();

        assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
        assert_eq!(mutations[0].key, json!("01020304"));
        let game = vm.get_entity_state(0, &json!("01020304")).unwrap();
        assert_eq!(game["id"]["game_id"], json!("01020304"));
        assert_eq!(game["stats"]["fee"], json!(5));
        assert_eq!(game["stats"].get("started_slot"), None);
        // The hook's writes are not kept anywhere else either.
        assert_eq!(vm.get_entity_state(0, &json!(null)), None);
    }

    mod fingerprint {
        use super::super::{EntityBytecode, MultiEntityBytecode, OpCode};
        use crate::ast::FieldPath;
        use serde_json::json;
        use std::collections::{HashMap, HashSet};

        fn sample_handler() -> Vec<OpCode> {
            vec![
                OpCode::LoadEventField {
                    path: FieldPath::new(&["id", "mint"]),
                    dest: 0,
                    default: Some(json!({"b": 2, "a": 1})),
                },
                OpCode::CreateObject { dest: 1 },
                OpCode::SetField {
                    object: 1,
                    path: "state.volume".to_string(),
                    value: 0,
                },
                OpCode::EmitMutation {
                    entity_name: "PumpfunToken".to_string(),
                    key: 0,
                    state: 1,
                },
            ]
        }

        fn bytecode(handler_insert_order: &[&str]) -> MultiEntityBytecode {
            let mut handlers = HashMap::new();
            for event_type in handler_insert_order {
                handlers.insert(event_type.to_string(), sample_handler());
            }
            let mut entities = HashMap::new();
            entities.insert(
                "PumpfunToken".to_string(),
                EntityBytecode {
                    state_id: 0,
                    handlers,
                    entity_name: "PumpfunToken".to_string(),
                    when_events: HashSet::from(["pump::BuyIxState".to_string()]),
                    non_emitted_fields: HashSet::new(),
                    computed_paths: vec!["state.market_cap".to_string()],
                    computed_fields_evaluator: None,
                },
            );
            let mut event_routing = HashMap::new();
            event_routing.insert(
                "pump::TokenState".to_string(),
                vec!["PumpfunToken".to_string()],
            );
            MultiEntityBytecode {
                entities,
                event_routing,
                when_events: HashSet::from(["pump::BuyIxState".to_string()]),
                proto_router: crate::proto_router::ProtoRouter::new(),
            }
        }

        #[test]
        fn stable_across_recomputation_and_insertion_order() {
            let a = bytecode(&["pump::TokenState", "pump::BuyIxState"]);
            let b = bytecode(&["pump::BuyIxState", "pump::TokenState"]);
            assert_eq!(a.fingerprint(), a.fingerprint());
            assert_eq!(a.fingerprint(), b.fingerprint());
            assert_eq!(a.fingerprint().len(), 64);
        }

        #[test]
        fn changes_when_an_opcode_operand_changes() {
            let base = bytecode(&["pump::TokenState"]);
            let mut modified = bytecode(&["pump::TokenState"]);
            let entity = modified.entities.get_mut("PumpfunToken").unwrap();
            match &mut entity.handlers.get_mut("pump::TokenState").unwrap()[2] {
                OpCode::SetField { path, .. } => *path = "state.volume_usd".to_string(),
                other => panic!("unexpected opcode: {other:?}"),
            }
            assert_ne!(base.fingerprint(), modified.fingerprint());
        }

        #[test]
        fn changes_when_routing_changes() {
            let base = bytecode(&["pump::TokenState"]);
            let mut modified = bytecode(&["pump::TokenState"]);
            modified.event_routing.insert(
                "pump::SellIxState".to_string(),
                vec!["PumpfunToken".to_string()],
            );
            assert_ne!(base.fingerprint(), modified.fingerprint());
        }

        /// Golden hash: this value changing means every deployed snapshot gets
        /// invalidated (bytecode_hash mismatch -> cold start). That can be a
        /// legitimate consequence of a semantic change to the sample opcodes or
        /// to the canonical encoding itself — but it must be a *decision*, not
        /// an accident. Update the constant only when snapshot invalidation is
        /// intended.
        #[test]
        fn golden_hash_pins_the_canonical_encoding() {
            let expected = "acf7ea8bf39ca830a9e6c2b2d2fb471da68e57e24bbb4c256964a42a498f75f2";
            assert_eq!(bytecode(&["pump::TokenState"]).fingerprint(), expected);
        }
    }
}