interpretthis 0.4.1

// Copyright 2026 Thomas Santerre and Moderately AI Inc.
//
// SPDX-License-Identifier: MIT OR Apache-2.0

use std::{sync::Arc, time::Instant};

use rustc_hash::FxHashMap;
use rustpython_parser::ast::{Expr, Stmt};
use tokio::sync::Semaphore;

use crate::{
    config::InterpreterConfig,
    value::{ClassValue, Value},
};

/// Mutable cursor for a [`crate::value::Value::BuiltinIter`] — the
/// infinite `itertools` producers. Each `__next__` advances it in place.
#[derive(Debug, Clone)]
pub enum BuiltinIterState {
    /// `itertools.count(start, step)` — yields `start, start+step, ...`.
    /// Holds `Value`s so a float `start`/`step` counts like CPython.
    Count { next: crate::value::Value, step: crate::value::Value },
    /// `itertools.cycle(iterable)` — repeats a buffered sequence forever
    /// (empty buffer is immediately exhausted).
    Cycle { items: Vec<crate::value::Value>, pos: usize },
    /// `itertools.repeat(obj)` with no count — yields `obj` forever.
    Repeat { value: crate::value::Value },
    /// `iter(list)` — holds the shared list and a cursor, so the iterator
    /// reflects mutations made before the cursor reaches them (CPython's
    /// `list_iterator` semantics), unlike an eager snapshot.
    ListIter { list: crate::value::SharedList, index: usize },
    /// `iter(bytearray)` — the shared-buffer counterpart of `ListIter`, yielding
    /// each byte as an int.
    BytearrayIter { data: crate::value::SharedByteArray, index: usize },
}

/// `itertools.count` step: `next + step` over the numeric types count
/// accepts (int/bigint/bool stay exact and promote past i64; any float
/// operand yields a float, matching CPython).
fn count_add(next: &Value, step: &Value) -> Value {
    use num_bigint::BigInt;
    let as_big = |v: &Value| -> Option<BigInt> {
        match v {
            Value::Int(n) => Some(BigInt::from(*n)),
            Value::BigInt(b) => Some((**b).clone()),
            Value::Bool(b) => Some(BigInt::from(i64::from(*b))),
            _ => None,
        }
    };
    let as_f64 = |v: &Value| -> Option<f64> {
        match v {
            Value::Float(f) => Some(*f),
            Value::Int(n) => Some(*n as f64),
            Value::Bool(b) => Some(f64::from(u8::from(*b))),
            Value::BigInt(b) => {
                Some(num_traits::ToPrimitive::to_f64(&**b).unwrap_or(f64::INFINITY))
            }
            _ => None,
        }
    };
    match (as_big(next), as_big(step)) {
        (Some(a), Some(b)) => crate::value::int_from_bigint(a + b),
        _ => match (as_f64(next), as_f64(step)) {
            (Some(a), Some(b)) => Value::Float(a + b),
            // Non-numeric operands can't occur (make_count validates),
            // but degrade gracefully rather than panic.
            _ => next.clone(),
        },
    }
}

/// Suspended generator function frame (true yield/resume, not eager buffer).
#[derive(Debug, Clone)]
pub struct GeneratorFrame {
    #[allow(dead_code)]
    pub func_name: String,
    pub source: String,
    /// Body AST (shared with function_bodies cache).
    pub body: std::sync::Arc<Vec<rustpython_parser::ast::Stmt>>,
    /// Names this frame may touch (for capture/restore).
    pub touched: Vec<String>,
    /// Frame-local bindings at last suspend (or initial args).
    pub locals: rustc_hash::FxHashMap<String, crate::value::Value>,
    pub started: bool,
    pub finished: bool,
    pub closed: bool,
    /// Result of the `yield` expression on resume (`send` value).
    pub send_value: crate::value::Value,
    /// When true, the next `yield` expression returns `send_value` instead of suspending.
    pub resume_at_yield: bool,
    /// Exception injected by `generator.throw(...)`: raised at the suspended
    /// `yield` on resume (so the generator's own `try/except` can catch it)
    /// instead of the yield returning `send_value`.
    pub pending_throw: Option<Box<crate::value::ExceptionValue>>,
    /// Next top-level statement index in `body`.
    pub stmt_index: usize,
    /// Nested for-loop / yield-from resume states (innermost last).
    pub for_stack: Vec<GeneratorForState>,
    /// Resume positions for suspended `while` loops, keyed by the loop's AST
    /// node byte-offset (stable across resumes since it is the same parsed
    /// node). Each entry `(offset, i)` means "re-enter that while's body at
    /// statement `i` without re-checking the condition". A stack rather than a
    /// single slot so a `while` nested in another `while`
    /// (`while a: while b: yield`) keeps both live — analogous to `for_stack`'s
    /// per-target keying.
    pub while_resume: Vec<(u32, usize)>,
    /// Resume positions for the stack of nested `try` blocks the
    /// generator is suspended inside. `try_stack[d]` is the position at
    /// nesting depth `d` (outermost first): which phase (body / matched
    /// `except` handler / `else` / `finally`) and the statement index
    /// within it. A `yield` anywhere in any nesting level thus resumes
    /// exactly where it suspended rather than re-running the statement.
    /// Empty means no try is mid-suspension.
    pub try_stack: Vec<TryResume>,
    /// Resume positions for the stack of `if` statements the generator is
    /// suspended inside (outermost first, LIFO like `try_stack`). Lets a yield
    /// buried in an `if` branch — even after side-effecting statements — resume
    /// exactly at the yield without re-running the branch or re-evaluating the
    /// condition. Empty when no `if` is mid-suspension.
    pub if_stack: Vec<IfResume>,
    /// Resume state for the stack of `with` statements the generator is
    /// suspended inside (outermost first, LIFO). Carries the already-entered
    /// context managers so `__exit__` runs at the real block exit — not on
    /// each `yield` — which is what lets a `@contextmanager` generator yield
    /// from inside a `with`. Empty when no `with` is mid-suspension.
    pub with_stack: Vec<WithResume>,
    /// Value a delegated (`yield from`) sub-generator returned, captured when
    /// the sub-generator was drained on the first pass and handed back as the
    /// value of the `yield from` expression when this frame resumes past it.
    pub yield_from_return: Option<crate::value::Value>,
    /// While this frame is suspended inside `yield from <generator>`, the id of
    /// the sub-generator it is delegating to. `next`/`send`/`throw`/`close` are
    /// forwarded to it (lazily, so the sub stays suspended and its `finally`
    /// runs at the right time) until the sub is exhausted. `None` otherwise.
    pub delegating_to: Option<u64>,
}

/// Where a suspended generator is inside a top-level `try` statement.
#[derive(Debug, Clone, Copy)]
pub struct TryResume {
    pub phase: TryPhase,
    /// Statement index within the current phase's body.
    pub index: usize,
}

/// Where a suspended generator is inside a `with` statement, plus the
/// context managers it already entered (so `__exit__` can run in reverse
/// order at the real block exit).
#[derive(Debug, Clone)]
pub struct WithResume {
    /// Statement index within the `with` body to resume at.
    pub index: usize,
    /// The entered context-manager objects, in declaration order.
    pub managers: Vec<crate::value::Value>,
}

/// Where a suspended generator is inside an `if` statement.
#[derive(Debug, Clone, Copy)]
pub struct IfResume {
    /// `true` when suspended in the `else` branch, `false` in the `then` branch.
    pub in_orelse: bool,
    /// Statement index within that branch.
    pub index: usize,
}

/// The four `try` phases a `yield` can suspend inside.
#[derive(Debug, Clone, Copy)]
pub enum TryPhase {
    Body,
    /// The `except` handler at this index matched and is running.
    Handler(usize),
    Orelse,
    Finally,
}

/// Resume state for a `for` loop that suspended on `yield` in its body.
#[derive(Debug, Clone)]
pub struct GeneratorForState {
    pub items: Arc<Vec<crate::value::Value>>,
    /// Index of the current item (the one whose body is in progress).
    pub pos: usize,
    /// Next statement index in the for-body after a yield resume.
    pub body_index: usize,
    /// Simple name target only (`for x in ...`). Empty => yield-from drain.
    pub target: String,
    /// When the loop iterates a *lazy* source (a generator / lazy iterator),
    /// this holds the source so the loop pulls one item at a time instead of
    /// materialising it (which would drain — and hang on — an infinite source).
    /// `None` for materialised sources, which use `items`/`pos`.
    pub lazy_source: Option<crate::value::Value>,
    /// The item currently being processed on the lazy path — remembered across
    /// a `yield` so the resume continues the same item rather than pulling a new
    /// one (the source's own cursor already tracks position).
    pub current_item: Option<crate::value::Value>,
}

/// Internal state of the interpreter — variables, print buffer, limits.
pub struct InterpreterState {
    pub variables: FxHashMap<String, Value>,
    /// User-defined classes, keyed by class name. A `Value::Class` variable is
    /// just a handle naming an entry here; methods and class attributes live in
    /// the [`ClassValue`] so instances stay cheap and methods are stored once.
    pub classes: FxHashMap<String, ClassValue>,
    pub print_buffer: String,
    /// Active `contextlib.redirect_stdout` targets (innermost last). While
    /// non-empty, `append_print` writes to the top StringIO instead of
    /// `print_buffer`. Transient with-block resource (not persisted).
    pub stdout_redirects: Vec<crate::value::SharedStringIo>,
    /// Parse cache for user-function body AST nodes, keyed by name.
    /// `rustpython_parser::ast` is not `Serialize`, so the canonical source
    /// of truth is `FunctionDef::source`; this map is populated at
    /// definition time and re-populated from `source` on state import so
    /// hot-path calls don't re-parse. Bodies are stored behind `Arc` so
    /// `call_user_function` clones a pointer rather than the full AST
    /// vector per call — every recursive frame would otherwise hold a
    /// freshly cloned body across `execute_body(...).await`, scaling
    /// per-frame heap/stack cost with function body size.
    pub function_bodies: FxHashMap<String, Arc<Vec<Stmt>>>,
    /// Stores lambda body AST nodes, keyed by a unique id. `Arc` for
    /// the same reason as `function_bodies`.
    pub lambda_bodies: FxHashMap<String, Arc<Expr>>,
    /// The current source code being executed (used to extract function source).
    pub current_source: String,
    /// Per-body source stack. Pushed by `call_user_function` /
    /// `call_lambda` before executing the body, popped after. When
    /// non-empty, `eval_stmt`'s line-stamp uses the top of the stack
    /// instead of `current_source` — so an error inside a function
    /// body persisted from a prior `execute()` call points at the
    /// function definition's source line, not at line 1 of the
    /// current `execute()` call's source.
    pub body_source_stack: Vec<String>,
    /// The lexical `__qualname__` prefix stack (innermost last). A class body
    /// pushes the class's own qualname (`Outer.Inner`); a function/lambda call
    /// pushes `<qualname>.<locals>` while its body runs. A `def`/`class`/`lambda`
    /// evaluated in that context reads the top entry to form its own qualname
    /// (`Outer.method`, `outer.<locals>.inner`). Transient — only non-empty
    /// mid-definition/mid-call, never persisted.
    pub qualname_stack: Vec<String>,
    /// `random` module RNG — CPython's Mersenne Twister, seeded on first use
    /// (and re-seeded by `random.seed`) so seeded sequences match CPython
    /// bit-for-bit.
    pub random_state: crate::eval::modules::random_mod::MtState,
    pub operations_count: u64,
    /// Wall-clock start time for execution timeout tracking.
    pub execution_start: Instant,
    /// Active `decimal` context precision (CPython default 28).
    pub decimal_prec: i64,
    /// Approximate memory used by interpreter state, in bytes.
    pub memory_used_bytes: usize,
    /// Current nested call depth. Bumped on entry to each user function
    /// / lambda frame and decremented on exit. Guarded against
    /// `config.max_recursion_depth`.
    pub call_depth: u32,
    /// Current nested *expression* depth on the async eval path (recursive
    /// non-bracketed chains like `a.b.c…`, `not not…`, `"a"+"a"+…`, `f()()…`).
    /// Guarded against `config.max_recursion_depth` so a pathological chain
    /// raises a catchable RecursionError — and stops growing the host stack —
    /// instead of recursing to the op-count limit. This is *per call frame*:
    /// entering a function body saves and resets it (see `enter_call`), so it
    /// bounds expression nesting within one frame without conflating with the
    /// separate function-call recursion limit (`call_depth`).
    pub expr_depth: u32,
    /// Saved `expr_depth` values, one per active call frame, restored on
    /// `exit_call`. Keeps a caller's mid-expression depth intact across a call.
    pub saved_expr_depth: Vec<u32>,
    /// Arc pointers of the containers currently being rendered by
    /// `eval::render` (CPython's `Py_ReprEnter` reentrancy set). A container
    /// re-entered while already here is a cycle and renders as `[...]` / `{...}`.
    /// Lives in state (not a thread-local) because `render` is async and its
    /// task can migrate worker threads across an await. Transient — not
    /// serialized.
    pub repr_active: rustc_hash::FxHashSet<usize>,
    /// Method-call frame stack. Pushed by `call_method` on entry,
    /// popped on exit. Carries the defining class for the executing
    /// method plus the current `self`, so zero-arg `super()` can pick
    /// up both without threading them through every helper. Transient
    /// execution state — not part of the serialized checkpoint.
    pub method_frame_stack: Vec<MethodFrame>,
    /// Yield-buffer stack for generator functions (Track C). Each
    /// entry is the buffer of yielded values for one generator
    /// frame; pushed on entry to a generator body, popped on exit and
    /// wrapped as a `Value::List`. The stack lets nested generator
    /// calls keep their yields separate. Transient — not serialized.
    pub yield_stack: Vec<Vec<Value>>,
    /// Active suspended generators keyed by `Value::Generator::id`.
    pub generators: rustc_hash::FxHashMap<u64, GeneratorFrame>,
    /// Stack of generators currently being stepped (supports yield-from).
    pub active_generator_stack: Vec<u64>,
    /// Active-exception stack for exception-handler bodies. Pushed by
    /// `try_match_handlers` on entry to a matching handler, popped on
    /// exit. Bare `raise` re-raises the top; a new exception raised
    /// inside a handler picks the top as its implicit
    /// `__context__` (CPython's "during handling of the above
    /// exception, another exception occurred" chaining). Transient —
    /// not part of the serialized checkpoint.
    pub active_exception_stack: Vec<crate::value::ExceptionValue>,
    /// Cursor positions for `Value::Lazy` iterators (generator
    /// results), keyed by the variant's `cursor_id`. A position of N
    /// means the next item to yield is `items[N]`; when N >= len,
    /// the iterator is exhausted.
    pub lazy_cursors: FxHashMap<u64, usize>,
    /// Monotonic id source for `Value::Lazy::cursor_id`. Bumped on
    /// each new generator-iterator allocation; never reused so two
    /// concurrent generators can't collide.
    pub next_cursor_id: u64,
    /// Cursor state for `Value::BuiltinIter` (the infinite `itertools`
    /// producers), keyed by the variant's `id`. Two handles sharing an
    /// `id` advance the same cursor, matching CPython iterator identity.
    pub builtin_iters: FxHashMap<u64, BuiltinIterState>,
    /// Monotonic id source for `Value::BuiltinIter::id`.
    pub next_builtin_iter_id: u64,
    /// Shared storage for `nonlocal`-bound variables, keyed by the
    /// owning `FunctionDef::nonlocal_cell_id`. Every call to the same
    /// function sees and updates the same cell, so `n += 1` inside an
    /// inner function persists across calls — matching CPython's
    /// reference semantics for nonlocal captures.
    pub nonlocal_cells: FxHashMap<u64, FxHashMap<String, crate::value::Value>>,
    /// Monotonic id source for `FunctionDef::nonlocal_cell_id`. Bumped
    /// each time a new function-with-nonlocal is defined.
    pub next_nonlocal_cell_id: u64,
    /// Per-frame "cells this frame owns" — pushed on entry to a
    /// user-function call and popped on exit. When a nested `def`
    /// declares `nonlocal x`, the enclosing frame registers `x` →
    /// (cell_id) here. Subsequent assignments to `x` in this frame
    /// flow through `set_variable` and write-through to
    /// `nonlocal_cells[cell_id][x]` so the inner sees the outer's
    /// live value on its next call (CPython binds inner's nonlocal
    /// to outer's actual cell object; this is the equivalent
    /// write-through). Transient — not part of the serialized
    /// checkpoint.
    pub frame_cell_owners: Vec<FxHashMap<String, u64>>,
    /// User-set attributes on function objects (`func.attr = value`),
    /// keyed by `FunctionDef::body_key` so every `Value::Function` clone
    /// of the same `def` shares one attribute namespace — matching
    /// CPython's per-function-object `__dict__`. Empty for the common
    /// case (functions carry no custom attributes). Not serialized:
    /// `body_key` is `#[serde(skip)]`, so custom attributes do not
    /// survive a state export/import (a documented limitation, like the
    /// by-name body-cache rebuild).
    pub function_attrs: FxHashMap<String, indexmap::IndexMap<String, crate::value::Value>>,
    pub config: InterpreterConfig,
    /// Shared semaphore for concurrent tool calls.
    pub tool_semaphore: Arc<Semaphore>,
}

/// One entry on `InterpreterState::method_frame_stack`. The
/// `defining_class` is the class on whose body the executing method was
/// defined — `super()` resumes from the next MRO slot after this one.
/// `self_value` is the bound receiver at frame-push time (captured so
/// zero-arg `super()` can construct a proxy without re-reading from
/// variables). `self_local_name` is the local variable name in the
/// current scope that holds `self`; after a `super().<method>(...)`
/// call mutates the instance, the result is written back to this
/// variable so subsequent statements in the calling method see the
/// updated state.
#[derive(Debug, Clone)]
pub struct MethodFrame {
    pub defining_class: String,
    pub self_value: Value,
    pub self_local_name: Option<String>,
}

impl InterpreterState {
    /// Compute the `__qualname__` for a `def`/`lambda`/`class` named `name`
    /// evaluated in the current lexical context: dotted onto the enclosing
    /// `qualname_stack` prefix (`Outer.method`, `outer.<locals>.inner`), or bare
    /// `name` at module scope.
    #[must_use]
    pub fn qualname_for(&self, name: &str) -> String {
        match self.qualname_stack.last() {
            Some(parent) => format!("{parent}.{name}"),
            None => name.to_string(),
        }
    }

    pub fn new(config: InterpreterConfig) -> Self {
        // CPython binds `__name__ = "__main__"` at module scope when a file is
        // executed directly; the standard `if __name__ == "__main__":` guard at
        // the bottom of every script depends on it. This interpreter
        // executes user code as a single top-level "module", so bind the same
        // default. `__name__` is not in DANGEROUS_NAMES or BLOCKED_ATTRIBUTES
        // (it's a benign dunder); `state_keys()` filters underscore-prefixed
        // names so it stays out of user-visible state listings.
        let mut variables = FxHashMap::default();
        variables.insert("__name__".to_string(), Value::String("__main__".into()));
        // Register `time.struct_time` as a namedtuple-shaped class so
        // `date.timetuple()` can build an Instance of it (indexing, `.tm_*`
        // attributes, and the `time.struct_time(...)` repr all reuse the
        // namedtuple `_fields` machinery). The dotted name can't be created via
        // the user-facing namedtuple() (invalid identifier), so it's seeded here.
        let mut classes = FxHashMap::default();
        let mut struct_time = crate::value::ClassValue::new("time.struct_time");
        struct_time.class_attrs.insert(
            "_fields".to_string(),
            Value::Tuple(
                [
                    "tm_year", "tm_mon", "tm_mday", "tm_hour", "tm_min", "tm_sec", "tm_wday",
                    "tm_yday", "tm_isdst",
                ]
                .into_iter()
                .map(|f| Value::String(f.into()))
                .collect(),
            ),
        );
        classes.insert("time.struct_time".to_string(), struct_time);
        // `datetime.IsoCalendarDate` for `date.isocalendar()` (year, week,
        // weekday), same namedtuple-shaped reuse.
        let mut iso = crate::value::ClassValue::new("datetime.IsoCalendarDate");
        iso.class_attrs.insert(
            "_fields".to_string(),
            Value::Tuple(
                ["year", "week", "weekday"].into_iter().map(|f| Value::String(f.into())).collect(),
            ),
        );
        classes.insert("datetime.IsoCalendarDate".to_string(), iso);
        // `functools.CacheInfo` for `lru_cache().cache_info()`.
        let mut cache_info = crate::value::ClassValue::new("CacheInfo");
        cache_info.class_attrs.insert(
            "_fields".to_string(),
            Value::Tuple(
                ["hits", "misses", "maxsize", "currsize"]
                    .into_iter()
                    .map(|f| Value::String(f.into()))
                    .collect(),
            ),
        );
        classes.insert("CacheInfo".to_string(), cache_info);
        // `decimal.DecimalTuple` for `Decimal.as_tuple()` (sign, digits, exponent).
        let mut dec_tuple = crate::value::ClassValue::new("DecimalTuple");
        dec_tuple.class_attrs.insert(
            "_fields".to_string(),
            Value::Tuple(
                ["sign", "digits", "exponent"]
                    .into_iter()
                    .map(|f| Value::String(f.into()))
                    .collect(),
            ),
        );
        classes.insert("DecimalTuple".to_string(), dec_tuple);
        Self {
            variables,
            classes,
            print_buffer: String::new(),
            stdout_redirects: Vec::new(),
            function_bodies: FxHashMap::default(),
            lambda_bodies: FxHashMap::default(),
            current_source: String::new(),
            body_source_stack: Vec::new(),
            qualname_stack: Vec::new(),
            random_state: crate::eval::modules::random_mod::MtState::new(),
            operations_count: 0,
            execution_start: Instant::now(),
            decimal_prec: 28,
            memory_used_bytes: 0,
            call_depth: 0,
            expr_depth: 0,
            saved_expr_depth: Vec::new(),
            repr_active: rustc_hash::FxHashSet::default(),
            method_frame_stack: Vec::new(),
            yield_stack: Vec::new(),
            generators: rustc_hash::FxHashMap::default(),
            active_generator_stack: Vec::new(),
            active_exception_stack: Vec::new(),
            lazy_cursors: FxHashMap::default(),
            next_cursor_id: 0,
            builtin_iters: FxHashMap::default(),
            next_builtin_iter_id: 0,
            nonlocal_cells: FxHashMap::default(),
            function_attrs: FxHashMap::default(),
            next_nonlocal_cell_id: 0,
            frame_cell_owners: Vec::new(),
            tool_semaphore: Arc::new(Semaphore::new(config.max_concurrent_tools as usize)),
            config,
        }
    }

    /// Bump the call-depth counter for entry to a new user function /
    /// lambda frame. Returns `RecursionLimitExceeded` when depth would
    /// exceed `config.max_recursion_depth` — the counter is not bumped
    /// on that failure path, so the caller must only pair a successful
    /// `enter_call` with a matching `exit_call`.
    pub fn enter_call(&mut self) -> Result<(), crate::error::InterpreterError> {
        if self.call_depth >= self.config.max_recursion_depth {
            return Err(crate::error::InterpreterError::RecursionLimitExceeded {
                limit: self.config.max_recursion_depth,
            });
        }
        self.call_depth = self.call_depth.saturating_add(1);
        // A function body is a fresh expression context: save the caller's
        // mid-expression depth and start the callee at zero, so `expr_depth`
        // never conflates with the (separately limited) call recursion.
        self.saved_expr_depth.push(self.expr_depth);
        self.expr_depth = 0;
        Ok(())
    }

    /// Decrement the call-depth counter on exit from a user frame and restore
    /// the caller's saved expression depth.
    pub fn exit_call(&mut self) {
        self.call_depth = self.call_depth.saturating_sub(1);
        if let Some(saved) = self.saved_expr_depth.pop() {
            self.expr_depth = saved;
        }
    }

    /// Bump the expression-nesting counter for a recursive async expression
    /// arm. Same limit and contract as [`enter_call`]: not bumped on the error
    /// path, so a successful `enter_expr` must be paired with one `exit_expr`.
    pub const fn enter_expr(&mut self) -> Result<(), crate::error::InterpreterError> {
        if self.expr_depth >= self.config.max_recursion_depth {
            return Err(crate::error::InterpreterError::RecursionLimitExceeded {
                limit: self.config.max_recursion_depth,
            });
        }
        self.expr_depth = self.expr_depth.saturating_add(1);
        Ok(())
    }

    /// Decrement the expression-nesting counter.
    pub const fn exit_expr(&mut self) {
        self.expr_depth = self.expr_depth.saturating_sub(1);
    }

    /// Set a variable, tracking memory usage.
    /// Returns an error if the memory limit is exceeded.
    ///
    /// Write-through to nonlocal cells: if the current frame owns a
    /// cell for this name (i.e. some inner function captured it as
    /// nonlocal), the new value is also stored in `nonlocal_cells`
    /// so the inner sees the up-to-date value on its next call.
    /// CPython binds inner's nonlocal to an actual cell object
    /// shared with outer's local; this write-through is the
    /// flat-state equivalent.
    pub fn set_variable(
        &mut self,
        name: &str,
        value: Value,
    ) -> Result<(), crate::error::InterpreterError> {
        // Release old value's memory if overwriting
        if let Some(old) = self.variables.get(name) {
            let old_size = estimate_value_size(old);
            self.memory_used_bytes = self.memory_used_bytes.saturating_sub(old_size);
        }
        // Track new value's memory
        let new_size = estimate_value_size(&value);
        self.memory_used_bytes = self.memory_used_bytes.saturating_add(new_size);

        // Write-through to any inner's nonlocal cell that this frame owns.
        if let Some(owners) = self.frame_cell_owners.last() {
            if let Some(&cell_id) = owners.get(name) {
                self.nonlocal_cells
                    .entry(cell_id)
                    .or_default()
                    .insert(name.to_string(), value.clone());
            }
        }

        self.variables.insert(name.to_string(), value);
        // Check memory immediately for large values
        self.check_memory()
    }

    #[inline]
    pub fn get_variable(&self, name: &str) -> Option<&Value> {
        self.variables.get(name)
    }

    /// Register a builtin lazy iterator, returning its `Value` handle.
    pub fn alloc_builtin_iter(
        &mut self,
        kind: crate::value::BuiltinIterName,
        st: BuiltinIterState,
    ) -> Value {
        let id = self.next_builtin_iter_id;
        self.next_builtin_iter_id = self.next_builtin_iter_id.wrapping_add(1);
        self.builtin_iters.insert(id, st);
        Value::BuiltinIter { id, kind }
    }

    /// Wrap eagerly-computed `items` in a one-shot `Lazy` iterator with a fresh
    /// cursor. Used by the builtins and itertools producers that CPython
    /// exposes as single-use iterators (`next()` advances them, a second pass
    /// sees only the remainder, they are neither subscriptable nor sized).
    pub fn alloc_lazy(&mut self, items: Vec<Value>) -> Value {
        self.alloc_lazy_kind(items, crate::value::LazyKind::Generator)
    }

    /// [`Self::alloc_lazy`] tagged with the producing iterator's kind so
    /// `type(x).__name__` / `repr` surface CPython's distinct iterator type
    /// (`map`, `filter`, `chain`, ...).
    pub fn alloc_lazy_kind(&mut self, items: Vec<Value>, kind: crate::value::LazyKind) -> Value {
        let cursor_id = self.next_cursor_id;
        self.next_cursor_id = self.next_cursor_id.wrapping_add(1);
        self.lazy_cursors.insert(cursor_id, 0);
        Value::Lazy { items, cursor_id, kind }
    }

    /// Advance a builtin lazy iterator by one. Returns `None` only when
    /// exhausted (an empty `cycle`) or the id is unknown.
    pub fn step_builtin_iter(&mut self, id: u64) -> Option<Value> {
        match self.builtin_iters.get_mut(&id)? {
            BuiltinIterState::Count { next, step } => {
                let v = next.clone();
                *next = count_add(next, step);
                Some(v)
            }
            BuiltinIterState::Cycle { items, pos } => {
                if items.is_empty() {
                    return None;
                }
                let v = items[*pos].clone();
                *pos = (*pos + 1) % items.len();
                Some(v)
            }
            BuiltinIterState::Repeat { value } => Some(value.clone()),
            BuiltinIterState::ListIter { list, index } => {
                let guard = list.lock();
                if *index < guard.len() {
                    let v = guard[*index].clone();
                    *index += 1;
                    Some(v)
                } else {
                    None
                }
            }
            BuiltinIterState::BytearrayIter { data, index } => {
                let guard = data.lock();
                if *index < guard.len() {
                    let v = Value::Int(i64::from(guard[*index]));
                    *index += 1;
                    Some(v)
                } else {
                    None
                }
            }
        }
    }

    pub fn delete_variable(&mut self, name: &str) -> Result<(), crate::error::InterpreterError> {
        match self.variables.remove(name) {
            Some(old) => {
                self.memory_used_bytes =
                    self.memory_used_bytes.saturating_sub(estimate_value_size(&old));
                Ok(())
            }
            None => Err(crate::error::InterpreterError::name_not_defined(name)),
        }
    }

    /// Returns user-visible state keys (excludes internal keys starting with _).
    pub fn state_keys(&self) -> Vec<String> {
        self.variables.keys().filter(|k| !k.starts_with('_')).cloned().collect()
    }

    pub fn clear_print_buffer(&mut self) {
        self.print_buffer.clear();
    }

    pub fn append_print(&mut self, text: &str) -> Result<(), crate::error::InterpreterError> {
        // Under an active `redirect_stdout`, print writes to that StringIO's
        // buffer (at its cursor) instead of the real stdout buffer.
        if let Some(target) = self.stdout_redirects.last() {
            let mut g = target.lock();
            let insert_at = g.pos.min(g.buf.chars().count());
            if insert_at >= g.buf.chars().count() {
                g.buf.push_str(text);
            } else {
                let mut chars: Vec<char> = g.buf.chars().collect();
                for (i, c) in text.chars().enumerate() {
                    if insert_at + i < chars.len() {
                        chars[insert_at + i] = c;
                    } else {
                        chars.push(c);
                    }
                }
                g.buf = chars.into_iter().collect();
            }
            g.pos = insert_at + text.chars().count();
            return Ok(());
        }
        let new_len = self.print_buffer.len() + text.len();
        // On 32-bit targets, max_stdout_bytes may exceed usize::MAX; saturate
        // to usize::MAX so the comparison stays meaningful (new_len is a usize
        // that cannot exceed usize::MAX regardless).
        let max_stdout = usize::try_from(self.config.max_stdout_bytes).unwrap_or(usize::MAX);
        if new_len > max_stdout {
            return Err(crate::error::InterpreterError::LimitExceeded(format!(
                "print output ({new_len} bytes) exceeds limit ({} bytes)",
                self.config.max_stdout_bytes
            )));
        }
        self.print_buffer.push_str(text);
        self.track_allocation(text.len())
    }

    pub const fn reset_operations(&mut self) {
        self.operations_count = 0;
    }

    pub fn increment_ops(&mut self) -> Result<(), crate::error::InterpreterError> {
        self.operations_count += 1;
        if self.operations_count >= self.config.max_operations {
            return Err(crate::error::InterpreterError::LimitExceeded(format!(
                "exceeded maximum of {} operations",
                self.config.max_operations
            )));
        }
        // Periodic checks every 100 ops (memory and timeout)
        if self.operations_count % 100 == 0 {
            self.check_memory()?;
            if let Some(max_time) = self.config.max_execution_time {
                if self.execution_start.elapsed() > max_time {
                    return Err(crate::error::InterpreterError::LimitExceeded(format!(
                        "execution time exceeded {max_time:?}"
                    )));
                }
            }
        }
        Ok(())
    }

    /// Check if memory usage exceeds the configured limit.
    pub fn check_memory(&self) -> Result<(), crate::error::InterpreterError> {
        let max_memory = usize::try_from(self.config.max_memory_bytes).unwrap_or(usize::MAX);
        if self.memory_used_bytes > max_memory {
            return Err(crate::error::InterpreterError::LimitExceeded(format!(
                "memory usage ({} bytes) exceeds limit ({} bytes)",
                self.memory_used_bytes, self.config.max_memory_bytes
            )));
        }
        Ok(())
    }

    /// Track an allocation of `bytes` against the memory budget.
    /// Returns an error if the budget is exceeded.
    pub fn track_allocation(&mut self, bytes: usize) -> Result<(), crate::error::InterpreterError> {
        self.memory_used_bytes = self.memory_used_bytes.saturating_add(bytes);
        let max_memory = usize::try_from(self.config.max_memory_bytes).unwrap_or(usize::MAX);
        if self.memory_used_bytes > max_memory {
            Err(crate::error::InterpreterError::LimitExceeded(format!(
                "memory usage ({} bytes) exceeds limit ({} bytes)",
                self.memory_used_bytes, self.config.max_memory_bytes
            )))
        } else {
            Ok(())
        }
    }

    /// Release `bytes` from the memory budget (e.g., when a variable is deleted).
    pub const fn release_allocation(&mut self, bytes: usize) {
        self.memory_used_bytes = self.memory_used_bytes.saturating_sub(bytes);
    }
}

/// Per-`Value` enum slot footprint, including the discriminant. Set
/// from the boxed-exception layout (`size_of::<Value>() == 80` on the
/// supported 64-bit targets). Used by [`estimate_value_size`] to
/// account for the per-slot overhead in every container — without it,
/// a `[Value::Int(0); 1000]` reports 8_000 B (payload only) when its
/// actual heap footprint is dominated by the enum slots plus the `Vec`
/// header.
const VALUE_SLOT_BYTES: usize = 80;

/// `String` header footprint: pointer + length + capacity = 3 × `usize`
/// on a 64-bit target. `CompactString` matches the layout exactly —
/// SSO is in the header bits, not extra space — so the same constant
/// applies to both `String` and `CompactString` fields.
const STRING_HEADER_BYTES: usize = 24;

/// Per-entry hash bucket overhead in an `IndexMap`-style ordered map.
/// Approximate: each entry costs the key + value bytes plus a hash
/// slot (8 B) and a doubly-tracked-back-index slot (8 B). The bucket
/// table itself is sized by `IndexMap::capacity()`, which we don't
/// inspect; this per-entry constant rolls the amortised cost in.
const INDEXMAP_PER_ENTRY_BYTES: usize = 16;

/// Estimate the memory footprint of a `Value` in bytes. The result
/// includes per-`Value`-slot overhead, container headers, string
/// headers, and amortised hash-table per-entry costs so that
/// `Interpreter::accounted_bytes()` tracks within ~2× of true RSS for
/// typical workloads (vs the prior payload-only accounting that
/// under-reported by ~10×). The number gates `max_memory_bytes` in
/// the sandbox so honest accounting is load-bearing for resource
/// protection — under-reporting let snippets allocate ~10× their
/// configured budget before tripping the limit.
pub fn estimate_value_size(value: &crate::value::Value) -> usize {
    // A value can nest arbitrarily deep (`a = []; for _ in range(9999): a = [a]`
    // is loop-built, so it never trips the recursion limit), and this walk
    // recurses once per level. Grow the host stack on demand — the same crate
    // and reason as the evaluator — so a deep-but-finite value is sized rather
    // than overflowing the stack and aborting the process. The `try_lock`
    // arms already handle self-referential cycles.
    stacker::maybe_grow(EST_STACK_RED_ZONE, EST_STACK_GROW, || estimate_value_size_inner(value))
}

const EST_STACK_RED_ZONE: usize = 512 * 1024;
const EST_STACK_GROW: usize = 32 * 1024 * 1024;

#[expect(
    clippy::match_same_arms,
    reason = "match arms are grouped by variant family (numerics, sequences, mappings, Track D types) for readability; merging same-body arms would scatter them across the table"
)]
fn estimate_value_size_inner(value: &crate::value::Value) -> usize {
    use crate::value::Value;
    match value {
        Value::None | Value::NotImplemented | Value::Ellipsis => 0,
        Value::OperatorGetter(_) => 32,
        Value::Bool(_) => 1,
        // i64 and f64 are both 8 bytes.
        Value::Int(_) | Value::Float(_) => 8,
        // complex is a boxed pair of f64.
        Value::Complex(_) => 16,
        // Approximate limb storage for big integers.
        Value::BigInt(b) => 16 + (b.bits() as usize / 8).saturating_add(8),
        Value::String(s) => STRING_HEADER_BYTES + s.len(),
        Value::Bytes(b) => STRING_HEADER_BYTES + b.len(),
        Value::ByteArray(b) => STRING_HEADER_BYTES + b.lock().len(),
        Value::StringIO(io) => STRING_HEADER_BYTES + io.try_lock().map_or(0, |g| g.buf.len()),
        Value::MemoryView(inner) => 16 + estimate_value_size(inner),
        // List is shared via Arc<Mutex<Vec>>; lock to walk under the
        // guard. Tuple/Set wrap plain Vec<Value> and walk directly.
        // `try_lock` (not `lock`): the container is already locked either by an
        // enclosing mutating method (`l.append(l)` holds the receiver lock while
        // sizing the argument) or by this walk recursing into a self-reference
        // (the guard is held across the element walk). Either way a re-entrant
        // `lock` would deadlock; skip it — its slot is already counted by the
        // parent.
        Value::List(items) | Value::Array { items, .. } => {
            let Some(mut guard) = items.try_lock() else {
                return 0;
            };
            guard.cached_size(|items| {
                STRING_HEADER_BYTES
                    + items.len() * VALUE_SLOT_BYTES
                    + items.iter().map(estimate_value_size).sum::<usize>()
            })
        }
        Value::Tuple(items) => {
            STRING_HEADER_BYTES
                + items.len() * VALUE_SLOT_BYTES
                + items.iter().map(estimate_value_size).sum::<usize>()
        }
        // An open-addressing set table over-allocates (load factor ≤3/5, so
        // ~2× capacity) with a Value slot + i64 hash per slot; approximate that
        // plus the elements' own sizes.
        Value::Set(_) | Value::Frozenset(_) => {
            let items = value.set_items().unwrap_or_default();
            STRING_HEADER_BYTES
                + items.len() * 2 * (VALUE_SLOT_BYTES + std::mem::size_of::<i64>())
                + items.iter().map(estimate_value_size).sum::<usize>()
        }
        Value::Dict(map) | Value::OrderedDict(map) => {
            let Some(mut guard) = map.try_lock() else {
                return 0;
            };
            guard.cached_size(|map| {
                48 + map.len() * (INDEXMAP_PER_ENTRY_BYTES + VALUE_SLOT_BYTES)
                    + map
                        .iter()
                        .map(|(k, v)| estimate_key_size(k) + estimate_value_size(v))
                        .sum::<usize>()
            })
        }
        // `Function` and `Lambda` are `Arc<…>`-wrapped (see F2.5). Count each
        // Value::Function reference as a fixed pointer-plus-header weight,
        // **not** the recursive walk of the inner closure. Walking through
        // the Arc would reintroduce the O(2^N) over-count that motivated
        // the Arc switch: the same FunctionDef is reachable from many
        // closures, and the heap holds it once. The actual closure storage
        // is counted exactly once at construction time (eval_function_def),
        // through the closure entries it contains at that snapshot moment.
        Value::Function(_) | Value::Lambda(_) | Value::LazyProxy(_) => 64,
        Value::Range { .. } => 24,
        Value::Exception(e) => 32 + e.type_name.len() + e.message.len(),
        Value::ExceptionMethod { method, exception } => {
            32 + method.len() + exception.type_name.len() + exception.message.len()
        }
        // Type / class / module handles carry just a name.
        Value::Type(n) | Value::Class(n) | Value::Module(n) => 8 + n.len(),
        Value::ModuleFunction { module, name } => 16 + module.len() + name.len(),
        Value::Date(_) => 16,
        Value::ReMatch(m) => {
            16 + m.groups.iter().flatten().map(|g| g.text.len() + 16).sum::<usize>()
        }
        Value::RePattern(p) => 16 + p.len(),
        Value::Slice(_) => 32,
        Value::Instance(inst) => {
            let Some(fields) = inst.fields.try_lock() else {
                return 16 + inst.class_name.len();
            };
            16 + inst.class_name.len()
                + fields.iter().map(|(k, v)| k.len() + estimate_value_size(v)).sum::<usize>()
        }
        Value::Super { defining_class, instance } => {
            16 + defining_class.len() + estimate_value_size(&Value::Instance((**instance).clone()))
        }
        Value::SuperClass { defining_class, class_name } => {
            16 + defining_class.len() + class_name.len()
        }
        Value::Counter(map) => {
            48 + map.len() * (INDEXMAP_PER_ENTRY_BYTES + VALUE_SLOT_BYTES)
                + map
                    .iter()
                    .map(|(k, v)| estimate_key_size(k) + estimate_value_size(v))
                    .sum::<usize>()
        }
        // 24 bytes for DateTime (matches Range); 12 for Time; 8 each
        // for TimeDelta (i64 micros) and TimeZone (i32 secs). Inlined
        // rather than merged with Range / Int arms because the
        // grouping reads as "Track D variants" — keeping them
        // adjacent helps readers.
        Value::DateTime { .. } => 24,
        Value::Time(_) => 12,
        Value::TimeDelta(_) | Value::TimeZone(_) => 8,
        Value::HashDigest { algo, bytes } => 16 + algo.len() + bytes.len(),
        Value::Deque { items, .. } => {
            STRING_HEADER_BYTES
                + items.len() * VALUE_SLOT_BYTES
                + items.iter().map(estimate_value_size).sum::<usize>()
        }
        Value::DefaultDict(data) => {
            48 + estimate_value_size(&data.factory)
                + data.items.len() * (INDEXMAP_PER_ENTRY_BYTES + VALUE_SLOT_BYTES)
                + data
                    .items
                    .iter()
                    .map(|(k, v)| estimate_key_size(k) + estimate_value_size(v))
                    .sum::<usize>()
        }
        // The maps are shared Dict handles counted elsewhere; charge a
        // flat per-map pointer weight to avoid double-counting.
        Value::ChainMap(maps) => 24 + maps.len() * 8,
        Value::Template(t) => 24 + t.len(),
        // The dict is a shared handle counted elsewhere.
        Value::DictView { .. } => 16,
        Value::EnumMember { class_name, member_name, value, .. } => {
            32 + class_name.len() + member_name.len() + estimate_value_size(value)
        }
        // BigDecimal and BigRational allocate a digit vector on the
        // heap. We over-estimate at a fixed 48 (the largest practical
        // single-instance memory for typical inputs) rather than
        // recursing into the BigInt's digit count — the bound stays
        // tight for the common case (Decimal("3.14"), Fraction(3, 7))
        // and only loosens for very long decimals, which a Python
        // script under our sandbox is unlikely to construct.
        Value::Decimal(..) | Value::Fraction(_) => 48,
        // Bound-method snapshot: pointer-sized header + the captured
        // receiver's own footprint (which is itself bounded by the
        // memory limit since the receiver had to live in a variable).
        Value::BoundMethod { receiver, method } => {
            use crate::value::{BoundMethodReceiver, BoundMethodStep};
            let receiver_size = match receiver {
                BoundMethodReceiver::Snapshot(v) => estimate_value_size(v),
                BoundMethodReceiver::Place { root, steps } => {
                    root.len()
                        + steps
                            .iter()
                            .map(|s| match s {
                                BoundMethodStep::Index(v) => 8 + estimate_value_size(v),
                                BoundMethodStep::Attr(n) => 8 + n.len(),
                            })
                            .sum::<usize>()
                }
            };
            16 + method.len() + receiver_size
        }
        Value::BuiltinTypeMethod { type_name, method } => 16 + type_name.len() + method.len(),
        Value::BuiltinName(n) | Value::ToolName(n) | Value::ExceptionType(n) => 16 + n.len(),
        Value::UnboundClassMethod { class, method } => 16 + class.len() + method.len(),
        Value::Lazy { items, .. } => 24 + items.iter().map(estimate_value_size).sum::<usize>(),
        Value::Generator { .. } => 16,
        // Handle is a flat id + kind tag; cursor state lives in
        // `builtin_iters` and is accounted there.
        Value::BuiltinIter { .. } => 16,
        Value::Partial(data) => {
            16 + estimate_value_size(&data.func)
                + data.args.iter().map(estimate_value_size).sum::<usize>()
                + data.keywords.values().map(estimate_value_size).sum::<usize>()
        }
        Value::LruCache(data) => {
            16 + estimate_value_size(&data.func)
                + data.cache.lock().values().map(estimate_value_size).sum::<usize>()
        }
        Value::SingleDispatch(sd) => {
            16 + estimate_value_size(&sd.default)
                + sd.registry.lock().values().map(estimate_value_size).sum::<usize>()
        }
        // Two short names referencing a class's PropertyDef; the def itself
        // lives in the class registry, not here.
        Value::Property { class_name, name } => 16 + class_name.len() + name.len(),
        // The captured call arguments dominate a coroutine's size.
        Value::Coroutine(c) => {
            32 + c.args.iter().map(estimate_value_size).sum::<usize>()
                + c.kwargs.values().map(estimate_value_size).sum::<usize>()
        }
    }
}

/// Estimate the memory size of a `ValueKey` in bytes.
pub fn estimate_key_size(key: &crate::value::ValueKey) -> usize {
    // A nested tuple/frozenset key can be loop-built arbitrarily deep, like a
    // value; grow the stack so sizing it cannot overflow.
    stacker::maybe_grow(EST_STACK_RED_ZONE, EST_STACK_GROW, || estimate_key_size_inner(key))
}

fn estimate_key_size_inner(key: &crate::value::ValueKey) -> usize {
    use crate::value::ValueKey;
    match key {
        ValueKey::None | ValueKey::Ellipsis => 0,
        ValueKey::Bool(_) => 1,
        ValueKey::Int(_) | ValueKey::Float(_) => 8,
        ValueKey::Complex(..) => 16,
        ValueKey::BigInt(b) => 16 + (b.bits() as usize / 8).saturating_add(8),
        ValueKey::String(s) => s.len(),
        ValueKey::Tuple(items) | ValueKey::Frozenset(items) => {
            24 + items.iter().map(estimate_key_size).sum::<usize>()
        }
        ValueKey::Instance { value, .. } => 8 + estimate_value_size(value),
        ValueKey::Date(_) | ValueKey::Time(_) | ValueKey::TimeDelta(_) => 8,
        ValueKey::DateTime { .. } => 16,
        ValueKey::Decimal(_) | ValueKey::Fraction(_) => 48,
    }
}