diff --git a/contracts/split/src/batch_processor.rs b/contracts/split/src/batch_processor.rs
new file mode 100644
index 0000000..93c9347
--- /dev/null
+++ b/contracts/split/src/batch_processor.rs
@@ -0,0 +1,318 @@
+//! Batch transaction processing (issue #1016).
+//!
+//! Groups many payment-like items into validated batches that are processed
+//! in fixed-size chunks, so per-call overhead is paid once per chunk instead of
+//! once per item. Heap free and `no_std` friendly.
+//!
+//! - **Design** — [`BatchConfig`] bounds the batch, chunk and total amount.
+//! - **Validation** — [`validate`] rejects empty or oversized batches, bad
+//!   amounts, duplicate ids and totals that overflow or exceed the cap before
+//!   anything is processed.
+//! - **Error handling** — [`process`] keeps going after an item fails and
+//!   records each failure in the [`BatchReport`] instead of aborting.
+//! - **Monitoring & analytics** — [`BatchStats`] accumulates counters across
+//!   batches and derives success rate, average batch size and calls saved.
+
+/// Hard upper bound on items in one batch.
+pub const MAX_BATCH_ITEMS: usize = 64;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum BatchError {
+    /// `max_items`, `chunk_size` is zero or exceeds [`MAX_BATCH_ITEMS`].
+    InvalidConfig,
+    EmptyBatch,
+    TooManyItems,
+    /// An item amount is zero or negative.
+    InvalidAmount,
+    DuplicateId,
+    /// Summing amounts overflowed `i128`.
+    Overflow,
+    /// Total amount is above `max_total`.
+    TotalTooLarge,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct BatchConfig {
+    pub max_items: usize,
+    /// Items handled per chunk (one downstream call per chunk).
+    pub chunk_size: usize,
+    /// Maximum combined amount of a batch. `0` disables the cap.
+    pub max_total: i128,
+}
+
+impl BatchConfig {
+    fn check(&self) -> Result<(), BatchError> {
+        if self.max_items == 0
+            || self.max_items > MAX_BATCH_ITEMS
+            || self.chunk_size == 0
+            || self.chunk_size > MAX_BATCH_ITEMS
+        {
+            return Err(BatchError::InvalidConfig);
+        }
+        Ok(())
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct BatchItem {
+    pub id: u32,
+    pub amount: i128,
+}
+
+/// Validate a batch and return its total amount.
+pub fn validate(config: &BatchConfig, items: &[BatchItem]) -> Result<i128, BatchError> {
+    config.check()?;
+    if items.is_empty() {
+        return Err(BatchError::EmptyBatch);
+    }
+    if items.len() > config.max_items {
+        return Err(BatchError::TooManyItems);
+    }
+
+    let mut total: i128 = 0;
+    for (i, item) in items.iter().enumerate() {
+        if item.amount <= 0 {
+            return Err(BatchError::InvalidAmount);
+        }
+        if items[..i].iter().any(|other| other.id == item.id) {
+            return Err(BatchError::DuplicateId);
+        }
+        total = total.checked_add(item.amount).ok_or(BatchError::Overflow)?;
+    }
+    if config.max_total > 0 && total > config.max_total {
+        return Err(BatchError::TotalTooLarge);
+    }
+    Ok(total)
+}
+
+/// Number of chunks (downstream calls) needed for `items` items.
+pub fn chunk_count(items: usize, chunk_size: usize) -> usize {
+    if chunk_size == 0 {
+        0
+    } else {
+        items.div_ceil(chunk_size)
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct ItemFailure {
+    pub id: u32,
+    /// Handler-defined error code.
+    pub code: u32,
+}
+
+#[derive(Clone, Copy, Debug)]
+pub struct BatchReport {
+    pub total_amount: i128,
+    pub succeeded: u32,
+    pub failed: u32,
+    pub chunks: u32,
+    failures: [ItemFailure; MAX_BATCH_ITEMS],
+    failure_count: usize,
+}
+
+impl BatchReport {
+    pub fn failures(&self) -> &[ItemFailure] {
+        &self.failures[..self.failure_count]
+    }
+}
+
+/// Cumulative counters across batches.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct BatchStats {
+    pub batches: u32,
+    pub rejected_batches: u32,
+    pub items: u32,
+    pub succeeded: u32,
+    pub failed: u32,
+    pub chunks: u32,
+}
+
+impl BatchStats {
+    /// Share of processed items that succeeded, in basis points.
+    pub fn success_rate_bps(&self) -> u32 {
+        if self.items == 0 {
+            return 0;
+        }
+        (self.succeeded as u64 * 10_000 / self.items as u64) as u32
+    }
+
+    pub fn average_batch_size(&self) -> u32 {
+        if self.batches == 0 {
+            0
+        } else {
+            self.items / self.batches
+        }
+    }
+
+    /// Downstream calls avoided versus processing every item individually.
+    pub fn calls_saved(&self) -> u32 {
+        self.items.saturating_sub(self.chunks)
+    }
+}
+
+/// Validate and process `items`, calling `handler` for every item.
+///
+/// `handler` returns `Err(code)` to fail one item without aborting the batch.
+pub fn process<F>(
+    config: &BatchConfig,
+    items: &[BatchItem],
+    stats: &mut BatchStats,
+    mut handler: F,
+) -> Result<BatchReport, BatchError>
+where
+    F: FnMut(&BatchItem) -> Result<(), u32>,
+{
+    let total_amount = match validate(config, items) {
+        Ok(total) => total,
+        Err(e) => {
+            stats.rejected_batches += 1;
+            return Err(e);
+        }
+    };
+
+    let mut report = BatchReport {
+        total_amount,
+        succeeded: 0,
+        failed: 0,
+        chunks: chunk_count(items.len(), config.chunk_size) as u32,
+        failures: [ItemFailure { id: 0, code: 0 }; MAX_BATCH_ITEMS],
+        failure_count: 0,
+    };
+
+    for chunk in items.chunks(config.chunk_size) {
+        for item in chunk {
+            match handler(item) {
+                Ok(()) => report.succeeded += 1,
+                Err(code) => {
+                    report.failed += 1;
+                    report.failures[report.failure_count] = ItemFailure { id: item.id, code };
+                    report.failure_count += 1;
+                }
+            }
+        }
+    }
+
+    stats.batches += 1;
+    stats.items += items.len() as u32;
+    stats.succeeded += report.succeeded;
+    stats.failed += report.failed;
+    stats.chunks += report.chunks;
+    Ok(report)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn config() -> BatchConfig {
+        BatchConfig { max_items: 10, chunk_size: 4, max_total: 1_000 }
+    }
+
+    fn item(id: u32, amount: i128) -> BatchItem {
+        BatchItem { id, amount }
+    }
+
+    #[test]
+    fn validation_rejects_bad_batches() {
+        let c = config();
+        assert_eq!(validate(&c, &[]), Err(BatchError::EmptyBatch));
+        assert_eq!(validate(&c, &[item(1, 0)]), Err(BatchError::InvalidAmount));
+        assert_eq!(validate(&c, &[item(1, -4)]), Err(BatchError::InvalidAmount));
+        assert_eq!(validate(&c, &[item(1, 5), item(1, 6)]), Err(BatchError::DuplicateId));
+        assert_eq!(validate(&c, &[item(1, 600), item(2, 401)]), Err(BatchError::TotalTooLarge));
+
+        let many = [item(0, 1); 11];
+        assert_eq!(validate(&c, &many), Err(BatchError::TooManyItems));
+
+        let uncapped = BatchConfig { max_total: 0, ..c };
+        assert_eq!(
+            validate(&uncapped, &[item(1, i128::MAX), item(2, 1)]),
+            Err(BatchError::Overflow)
+        );
+    }
+
+    #[test]
+    fn validation_rejects_invalid_config() {
+        let zero_chunk = BatchConfig { chunk_size: 0, ..config() };
+        assert_eq!(validate(&zero_chunk, &[item(1, 1)]), Err(BatchError::InvalidConfig));
+        let huge = BatchConfig { max_items: MAX_BATCH_ITEMS + 1, ..config() };
+        assert_eq!(validate(&huge, &[item(1, 1)]), Err(BatchError::InvalidConfig));
+    }
+
+    #[test]
+    fn valid_batch_returns_total() {
+        assert_eq!(validate(&config(), &[item(1, 10), item(2, 20), item(3, 30)]), Ok(60));
+    }
+
+    #[test]
+    fn chunk_count_rounds_up() {
+        assert_eq!(chunk_count(0, 4), 0);
+        assert_eq!(chunk_count(1, 4), 1);
+        assert_eq!(chunk_count(4, 4), 1);
+        assert_eq!(chunk_count(5, 4), 2);
+        assert_eq!(chunk_count(9, 0), 0);
+    }
+
+    #[test]
+    fn failures_are_recorded_without_aborting() {
+        let mut stats = BatchStats::default();
+        let items = [item(1, 10), item(2, 20), item(3, 30), item(4, 40), item(5, 50)];
+        let mut seen = 0;
+        let report = process(&config(), &items, &mut stats, |it| {
+            seen += 1;
+            if it.id % 2 == 0 {
+                Err(it.id * 100)
+            } else {
+                Ok(())
+            }
+        })
+        .unwrap();
+
+        assert_eq!(seen, 5);
+        assert_eq!(report.total_amount, 150);
+        assert_eq!(report.succeeded, 3);
+        assert_eq!(report.failed, 2);
+        assert_eq!(report.chunks, 2);
+        assert_eq!(
+            report.failures(),
+            &[ItemFailure { id: 2, code: 200 }, ItemFailure { id: 4, code: 400 }]
+        );
+    }
+
+    #[test]
+    fn rejected_batch_runs_no_handler_and_is_counted() {
+        let mut stats = BatchStats::default();
+        let mut called = false;
+        let err = process(&config(), &[item(1, 0)], &mut stats, |_| {
+            called = true;
+            Ok(())
+        })
+        .err();
+        assert_eq!(err, Some(BatchError::InvalidAmount));
+        assert!(!called);
+        assert_eq!(stats.rejected_batches, 1);
+        assert_eq!(stats.batches, 0);
+    }
+
+    #[test]
+    fn stats_derive_analytics() {
+        let mut stats = BatchStats::default();
+        assert_eq!(stats.success_rate_bps(), 0);
+        assert_eq!(stats.average_batch_size(), 0);
+
+        let first = [item(1, 1), item(2, 1), item(3, 1), item(4, 1), item(5, 1), item(6, 1)];
+        process(&config(), &first, &mut stats, |_| Ok(())).unwrap();
+        let second = [item(1, 1), item(2, 1)];
+        process(&config(), &second, &mut stats, |it| if it.id == 2 { Err(1) } else { Ok(()) }).unwrap();
+
+        assert_eq!(stats.batches, 2);
+        assert_eq!(stats.items, 8);
+        assert_eq!(stats.succeeded, 7);
+        assert_eq!(stats.failed, 1);
+        assert_eq!(stats.chunks, 3);
+        assert_eq!(stats.success_rate_bps(), 8_750);
+        assert_eq!(stats.average_batch_size(), 4);
+        assert_eq!(stats.calls_saved(), 5);
+    }
+}
diff --git a/contracts/split/src/lazy_loader.rs b/contracts/split/src/lazy_loader.rs
new file mode 100644
index 0000000..b687c75
--- /dev/null
+++ b/contracts/split/src/lazy_loader.rs
@@ -0,0 +1,273 @@
+//! Lazy loading (issue #1018).
+//!
+//! A bounded read-through cache that loads values on demand, so only the
+//! entries actually touched are held in memory. Heap free and `no_std`
+//! friendly; time is expressed in ledger sequence numbers.
+//!
+//! - **On-demand loading** — [`LazyLoader::get_or_load`] calls the loader only
+//!   on a miss.
+//! - **Prefetching** — [`LazyLoader::prefetch`] warms keys that are about to be used.
+//! - **Cache invalidation** — entries expire after `ttl` ledgers, and can be
+//!   dropped with [`LazyLoader::invalidate`] / [`LazyLoader::invalidate_all`].
+//! - **Memory bound** — at most `CAP` entries are held; the least recently
+//!   used entry is evicted when full.
+//! - **Monitoring** — [`LoadMetrics`] counts hits, misses, loads and evictions.
+
+#[derive(Clone, Copy, Debug)]
+struct Slot<V: Copy> {
+    key: u64,
+    value: V,
+    loaded_at: u32,
+    last_used: u32,
+}
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct LoadMetrics {
+    pub hits: u32,
+    pub misses: u32,
+    pub loads: u32,
+    pub prefetched: u32,
+    pub evictions: u32,
+    pub invalidations: u32,
+    pub expirations: u32,
+}
+
+impl LoadMetrics {
+    /// Share of lookups served from memory, in basis points.
+    pub fn hit_rate_bps(&self) -> u32 {
+        let total = self.hits + self.misses;
+        if total == 0 {
+            0
+        } else {
+            (self.hits as u64 * 10_000 / total as u64) as u32
+        }
+    }
+}
+
+pub struct LazyLoader<V: Copy, const CAP: usize> {
+    slots: [Option<Slot<V>>; CAP],
+    /// Entries older than this many ledgers are reloaded. `0` disables expiry.
+    ttl: u32,
+    metrics: LoadMetrics,
+}
+
+impl<V: Copy, const CAP: usize> LazyLoader<V, CAP> {
+    pub fn new(ttl: u32) -> Self {
+        LazyLoader { slots: [None; CAP], ttl, metrics: LoadMetrics::default() }
+    }
+
+    pub fn metrics(&self) -> LoadMetrics {
+        self.metrics
+    }
+
+    /// Number of entries currently held in memory.
+    pub fn len(&self) -> usize {
+        self.slots.iter().filter(|s| s.is_some()).count()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    fn is_expired(&self, slot: &Slot<V>, now: u32) -> bool {
+        self.ttl > 0 && now.saturating_sub(slot.loaded_at) >= self.ttl
+    }
+
+    fn find(&self, key: u64) -> Option<usize> {
+        self.slots.iter().position(|s| matches!(s, Some(slot) if slot.key == key))
+    }
+
+    /// Index to store a new entry in: a free slot, else the least recently used.
+    fn victim(&mut self) -> usize {
+        if let Some(free) = self.slots.iter().position(|s| s.is_none()) {
+            return free;
+        }
+        let mut oldest = 0;
+        let mut oldest_used = u32::MAX;
+        for (i, s) in self.slots.iter().enumerate() {
+            if let Some(slot) = s {
+                if slot.last_used < oldest_used {
+                    oldest_used = slot.last_used;
+                    oldest = i;
+                }
+            }
+        }
+        self.metrics.evictions += 1;
+        oldest
+    }
+
+    fn store(&mut self, key: u64, value: V, now: u32) {
+        let index = match self.find(key) {
+            Some(i) => i,
+            None => self.victim(),
+        };
+        self.slots[index] = Some(Slot { key, value, loaded_at: now, last_used: now });
+    }
+
+    /// Return the cached value, loading it with `loader` on a miss or after expiry.
+    /// `None` from the loader means the value does not exist and nothing is cached.
+    pub fn get_or_load<F>(&mut self, key: u64, now: u32, loader: F) -> Option<V>
+    where
+        F: FnOnce(u64) -> Option<V>,
+    {
+        if let Some(i) = self.find(key) {
+            let slot = self.slots[i].expect("slot found above");
+            if self.is_expired(&slot, now) {
+                self.slots[i] = None;
+                self.metrics.expirations += 1;
+            } else {
+                self.metrics.hits += 1;
+                self.slots[i] = Some(Slot { last_used: now, ..slot });
+                return Some(slot.value);
+            }
+        }
+
+        self.metrics.misses += 1;
+        let value = loader(key)?;
+        self.metrics.loads += 1;
+        self.store(key, value, now);
+        Some(value)
+    }
+
+    /// Load every key in `keys` that is not already cached. Returns how many were loaded.
+    pub fn prefetch<F>(&mut self, keys: &[u64], now: u32, mut loader: F) -> u32
+    where
+        F: FnMut(u64) -> Option<V>,
+    {
+        let mut loaded = 0;
+        for &key in keys {
+            let fresh = match self.find(key) {
+                Some(i) => !self.is_expired(&self.slots[i].expect("slot found above"), now),
+                None => false,
+            };
+            if fresh {
+                continue;
+            }
+            if let Some(value) = loader(key) {
+                self.store(key, value, now);
+                self.metrics.loads += 1;
+                self.metrics.prefetched += 1;
+                loaded += 1;
+            }
+        }
+        loaded
+    }
+
+    /// Drop one entry. Returns `true` if it was cached.
+    pub fn invalidate(&mut self, key: u64) -> bool {
+        match self.find(key) {
+            Some(i) => {
+                self.slots[i] = None;
+                self.metrics.invalidations += 1;
+                true
+            }
+            None => false,
+        }
+    }
+
+    /// Drop every entry. Returns how many were cached.
+    pub fn invalidate_all(&mut self) -> usize {
+        let held = self.len();
+        self.slots = [None; CAP];
+        self.metrics.invalidations += held as u32;
+        held
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn square(key: u64) -> Option<u64> {
+        Some(key * key)
+    }
+
+    #[test]
+    fn loads_on_demand_and_serves_hits_from_memory() {
+        let mut cache: LazyLoader<u64, 4> = LazyLoader::new(0);
+        let mut calls = 0;
+        assert_eq!(cache.get_or_load(3, 1, |k| { calls += 1; square(k) }), Some(9));
+        assert_eq!(cache.get_or_load(3, 2, |k| { calls += 1; square(k) }), Some(9));
+        assert_eq!(calls, 1);
+        assert_eq!(cache.len(), 1);
+        let m = cache.metrics();
+        assert_eq!((m.hits, m.misses, m.loads), (1, 1, 1));
+        assert_eq!(m.hit_rate_bps(), 5_000);
+    }
+
+    #[test]
+    fn missing_values_are_not_cached() {
+        let mut cache: LazyLoader<u64, 2> = LazyLoader::new(0);
+        assert_eq!(cache.get_or_load(1, 0, |_| None), None);
+        assert!(cache.is_empty());
+        assert_eq!(cache.metrics().loads, 0);
+        assert_eq!(cache.metrics().misses, 1);
+    }
+
+    #[test]
+    fn memory_is_bounded_and_evicts_least_recently_used() {
+        let mut cache: LazyLoader<u64, 2> = LazyLoader::new(0);
+        cache.get_or_load(1, 1, square);
+        cache.get_or_load(2, 2, square);
+        // Touch key 1 so key 2 becomes the least recently used.
+        cache.get_or_load(1, 3, square);
+        cache.get_or_load(3, 4, square);
+
+        assert_eq!(cache.len(), 2);
+        assert_eq!(cache.metrics().evictions, 1);
+        let mut reloaded = false;
+        cache.get_or_load(2, 5, |k| { reloaded = true; square(k) });
+        assert!(reloaded, "key 2 should have been evicted");
+    }
+
+    #[test]
+    fn entries_expire_after_ttl() {
+        let mut cache: LazyLoader<u64, 2> = LazyLoader::new(10);
+        cache.get_or_load(5, 100, square);
+        let mut calls = 0;
+        cache.get_or_load(5, 109, |k| { calls += 1; square(k) });
+        assert_eq!(calls, 0);
+        cache.get_or_load(5, 110, |k| { calls += 1; square(k) });
+        assert_eq!(calls, 1);
+        assert_eq!(cache.metrics().expirations, 1);
+    }
+
+    #[test]
+    fn prefetch_warms_only_uncached_keys() {
+        let mut cache: LazyLoader<u64, 4> = LazyLoader::new(0);
+        cache.get_or_load(1, 0, square);
+        let mut loaded_keys = 0;
+        let n = cache.prefetch(&[1, 2, 3], 1, |k| { loaded_keys += 1; square(k) });
+        assert_eq!(n, 2);
+        assert_eq!(loaded_keys, 2);
+        assert_eq!(cache.metrics().prefetched, 2);
+
+        // Prefetched keys are now hits.
+        assert_eq!(cache.get_or_load(2, 2, |_| None), Some(4));
+        assert_eq!(cache.get_or_load(3, 2, |_| None), Some(9));
+        assert_eq!(cache.metrics().hits, 2);
+    }
+
+    #[test]
+    fn invalidation_forces_a_reload() {
+        let mut cache: LazyLoader<u64, 4> = LazyLoader::new(0);
+        cache.get_or_load(1, 0, square);
+        cache.get_or_load(2, 0, square);
+        assert!(cache.invalidate(1));
+        assert!(!cache.invalidate(1));
+        assert_eq!(cache.len(), 1);
+
+        assert_eq!(cache.invalidate_all(), 1);
+        assert!(cache.is_empty());
+        assert_eq!(cache.metrics().invalidations, 2);
+
+        let mut calls = 0;
+        cache.get_or_load(2, 1, |k| { calls += 1; square(k) });
+        assert_eq!(calls, 1);
+    }
+
+    #[test]
+    fn hit_rate_is_zero_without_lookups() {
+        assert_eq!(LoadMetrics::default().hit_rate_bps(), 0);
+    }
+}
diff --git a/contracts/split/src/parallel.rs b/contracts/split/src/parallel.rs
new file mode 100644
index 0000000..11c666a
--- /dev/null
+++ b/contracts/split/src/parallel.rs
@@ -0,0 +1,295 @@
+//! Parallel operation scheduling (issue #1019).
+//!
+//! Soroban executes a contract's operations sequentially, so this module does
+//! not spawn threads. It plans how a batch of operations *would* be split
+//! across independent lanes for an executor that can run lanes concurrently
+//! (an off-chain batcher, or several sub-invoices), and verifies that the plan
+//! is safe and the combined result is correct. Heap free and `no_std` friendly.
+//!
+//! - **Work distribution & load balancing** — [`schedule`] assigns each task
+//!   to the least-loaded lane.
+//! - **Synchronization** — tasks that share a `key` are pinned to one lane and
+//!   keep their relative order, so concurrent lanes never touch the same key.
+//! - **Result aggregation** — [`execute`] runs lane by lane and returns the
+//!   results in the original task order with checked totals.
+//! - **Metrics** — [`ScheduleMetrics`] reports speedup and balance.
+
+/// Maximum number of tasks per schedule.
+pub const MAX_TASKS: usize = 64;
+/// Maximum number of lanes.
+pub const MAX_LANES: usize = 8;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ParallelError {
+    /// `lanes` is zero or exceeds [`MAX_LANES`].
+    InvalidLanes,
+    /// More than [`MAX_TASKS`] tasks, or no tasks.
+    InvalidTaskCount,
+    /// A lane total overflowed `i128`.
+    Overflow,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct Task {
+    pub id: u32,
+    /// Tasks with the same key conflict and run on the same lane, in order.
+    pub key: u64,
+    /// Estimated cost, used for load balancing.
+    pub cost: u32,
+}
+
+#[derive(Clone, Copy, Debug)]
+pub struct Schedule {
+    lane_of: [u8; MAX_TASKS],
+    task_count: usize,
+    lane_cost: [u64; MAX_LANES],
+    lanes: usize,
+}
+
+impl Schedule {
+    pub fn lanes(&self) -> usize {
+        self.lanes
+    }
+
+    pub fn lane_of(&self, task_index: usize) -> Option<usize> {
+        if task_index < self.task_count {
+            Some(self.lane_of[task_index] as usize)
+        } else {
+            None
+        }
+    }
+
+    pub fn lane_cost(&self, lane: usize) -> u64 {
+        if lane < self.lanes {
+            self.lane_cost[lane]
+        } else {
+            0
+        }
+    }
+
+    pub fn metrics(&self) -> ScheduleMetrics {
+        let costs = &self.lane_cost[..self.lanes];
+        let total: u64 = costs.iter().sum();
+        let makespan = costs.iter().copied().max().unwrap_or(0);
+        let min_busy = costs.iter().copied().filter(|c| *c > 0).min().unwrap_or(0);
+        ScheduleMetrics {
+            tasks: self.task_count as u32,
+            total_cost: total,
+            makespan,
+            speedup_bps: if makespan == 0 { 0 } else { (total * 10_000 / makespan) as u32 },
+            imbalance_bps: if makespan == 0 {
+                0
+            } else {
+                ((makespan - min_busy) * 10_000 / makespan) as u32
+            },
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ScheduleMetrics {
+    pub tasks: u32,
+    /// Sum of every task's cost (the sequential runtime).
+    pub total_cost: u64,
+    /// Cost of the busiest lane (the parallel runtime).
+    pub makespan: u64,
+    /// `total_cost / makespan` in basis points; 10_000 means no speedup.
+    pub speedup_bps: u32,
+    /// Gap between the busiest and least busy lane, in basis points of the makespan.
+    pub imbalance_bps: u32,
+}
+
+/// Assign tasks to `lanes` lanes. Tasks sharing a key stay together; new keys
+/// go to the currently least-loaded lane (lowest index wins ties).
+pub fn schedule(tasks: &[Task], lanes: usize) -> Result<Schedule, ParallelError> {
+    if lanes == 0 || lanes > MAX_LANES {
+        return Err(ParallelError::InvalidLanes);
+    }
+    if tasks.is_empty() || tasks.len() > MAX_TASKS {
+        return Err(ParallelError::InvalidTaskCount);
+    }
+
+    let mut plan = Schedule {
+        lane_of: [0; MAX_TASKS],
+        task_count: tasks.len(),
+        lane_cost: [0; MAX_LANES],
+        lanes,
+    };
+
+    for (i, task) in tasks.iter().enumerate() {
+        let pinned = tasks[..i].iter().position(|earlier| earlier.key == task.key);
+        let lane = match pinned {
+            Some(earlier) => plan.lane_of[earlier] as usize,
+            None => {
+                let mut best = 0;
+                for l in 1..lanes {
+                    if plan.lane_cost[l] < plan.lane_cost[best] {
+                        best = l;
+                    }
+                }
+                best
+            }
+        };
+        plan.lane_of[i] = lane as u8;
+        plan.lane_cost[lane] += task.cost as u64;
+    }
+    Ok(plan)
+}
+
+/// Results of running a schedule.
+#[derive(Clone, Copy, Debug)]
+pub struct Execution {
+    results: [i128; MAX_TASKS],
+    count: usize,
+    lane_totals: [i128; MAX_LANES],
+    lanes: usize,
+    /// Sum of every result.
+    pub total: i128,
+}
+
+impl Execution {
+    /// Results in the original task order.
+    pub fn results(&self) -> &[i128] {
+        &self.results[..self.count]
+    }
+
+    pub fn lane_totals(&self) -> &[i128] {
+        &self.lane_totals[..self.lanes]
+    }
+}
+
+/// Run `work` for every task, one lane at a time, and aggregate the results.
+///
+/// Within a lane tasks run in their original order, so tasks sharing a key
+/// observe each other's effects exactly as in a sequential run.
+pub fn execute<F>(tasks: &[Task], plan: &Schedule, mut work: F) -> Result<Execution, ParallelError>
+where
+    F: FnMut(&Task) -> i128,
+{
+    let mut out = Execution {
+        results: [0; MAX_TASKS],
+        count: tasks.len(),
+        lane_totals: [0; MAX_LANES],
+        lanes: plan.lanes,
+        total: 0,
+    };
+
+    for lane in 0..plan.lanes {
+        for (i, task) in tasks.iter().enumerate() {
+            if plan.lane_of[i] as usize != lane {
+                continue;
+            }
+            let result = work(task);
+            out.results[i] = result;
+            out.lane_totals[lane] = out.lane_totals[lane]
+                .checked_add(result)
+                .ok_or(ParallelError::Overflow)?;
+        }
+    }
+
+    for lane in 0..plan.lanes {
+        out.total = out.total.checked_add(out.lane_totals[lane]).ok_or(ParallelError::Overflow)?;
+    }
+    Ok(out)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn task(id: u32, key: u64, cost: u32) -> Task {
+        Task { id, key, cost }
+    }
+
+    #[test]
+    fn rejects_invalid_inputs() {
+        let t = [task(1, 1, 1)];
+        assert_eq!(schedule(&t, 0).err(), Some(ParallelError::InvalidLanes));
+        assert_eq!(schedule(&t, MAX_LANES + 1).err(), Some(ParallelError::InvalidLanes));
+        assert_eq!(schedule(&[], 2).err(), Some(ParallelError::InvalidTaskCount));
+        let many = [task(0, 0, 1); MAX_TASKS + 1];
+        assert_eq!(schedule(&many, 2).err(), Some(ParallelError::InvalidTaskCount));
+    }
+
+    #[test]
+    fn independent_tasks_are_balanced_across_lanes() {
+        let tasks = [task(0, 0, 10), task(1, 1, 10), task(2, 2, 10), task(3, 3, 10)];
+        let plan = schedule(&tasks, 2).unwrap();
+        assert_eq!(plan.lane_cost(0), 20);
+        assert_eq!(plan.lane_cost(1), 20);
+
+        let m = plan.metrics();
+        assert_eq!(m.total_cost, 40);
+        assert_eq!(m.makespan, 20);
+        assert_eq!(m.speedup_bps, 20_000);
+        assert_eq!(m.imbalance_bps, 0);
+    }
+
+    #[test]
+    fn expensive_task_goes_alone_and_cheap_ones_fill_the_other_lane() {
+        let tasks = [task(0, 0, 100), task(1, 1, 10), task(2, 2, 10), task(3, 3, 10)];
+        let plan = schedule(&tasks, 2).unwrap();
+        assert_eq!(plan.lane_of(0), Some(0));
+        assert_eq!(plan.lane_of(1), Some(1));
+        assert_eq!(plan.lane_of(2), Some(1));
+        assert_eq!(plan.lane_of(3), Some(1));
+        assert_eq!(plan.lane_of(4), None);
+        assert_eq!(plan.metrics().makespan, 100);
+    }
+
+    #[test]
+    fn conflicting_tasks_share_a_lane() {
+        let tasks = [task(0, 7, 5), task(1, 8, 5), task(2, 7, 5), task(3, 9, 5), task(4, 7, 5)];
+        let plan = schedule(&tasks, 3).unwrap();
+        let lane = plan.lane_of(0).unwrap();
+        assert_eq!(plan.lane_of(2), Some(lane));
+        assert_eq!(plan.lane_of(4), Some(lane));
+        assert_eq!(plan.lane_cost(lane), 15);
+    }
+
+    #[test]
+    fn execution_matches_sequential_result_and_preserves_order() {
+        let tasks = [task(0, 1, 1), task(1, 2, 1), task(2, 1, 1), task(3, 3, 1), task(4, 2, 1)];
+
+        // Sequential reference: a per-key running counter.
+        let mut counters = [0i128; 4];
+        let mut expected = [0i128; 5];
+        for (i, t) in tasks.iter().enumerate() {
+            counters[t.key as usize] += 10;
+            expected[i] = counters[t.key as usize];
+        }
+
+        let plan = schedule(&tasks, 3).unwrap();
+        let mut parallel_counters = [0i128; 4];
+        let exec = execute(&tasks, &plan, |t| {
+            parallel_counters[t.key as usize] += 10;
+            parallel_counters[t.key as usize]
+        })
+        .unwrap();
+
+        assert_eq!(exec.results(), &expected);
+        assert_eq!(exec.total, expected.iter().sum::<i128>());
+        assert_eq!(exec.lane_totals().iter().sum::<i128>(), exec.total);
+        assert_eq!(exec.lane_totals().len(), 3);
+    }
+
+    #[test]
+    fn aggregation_reports_overflow() {
+        let tasks = [task(0, 1, 1), task(1, 2, 1)];
+        let plan = schedule(&tasks, 1).unwrap();
+        let err = execute(&tasks, &plan, |_| i128::MAX).err();
+        assert_eq!(err, Some(ParallelError::Overflow));
+
+        let plan = schedule(&tasks, 2).unwrap();
+        let err = execute(&tasks, &plan, |_| i128::MAX).err();
+        assert_eq!(err, Some(ParallelError::Overflow));
+    }
+
+    #[test]
+    fn single_lane_has_no_speedup() {
+        let tasks = [task(0, 1, 5), task(1, 2, 5)];
+        let plan = schedule(&tasks, 1).unwrap();
+        assert_eq!(plan.metrics().speedup_bps, 10_000);
+        assert_eq!(plan.lanes(), 1);
+    }
+}
diff --git a/contracts/split/src/storage_optimizer.rs b/contracts/split/src/storage_optimizer.rs
new file mode 100644
index 0000000..b5cea59
--- /dev/null
+++ b/contracts/split/src/storage_optimizer.rs
@@ -0,0 +1,299 @@
+//! Storage optimization (issue #1017).
+//!
+//! Helpers that shrink what a contract keeps in storage. Heap free and
+//! `no_std` friendly; callers pass the buffers.
+//!
+//! - **Analysis** — [`analyze`] summarizes entry counts and sizes.
+//! - **Compression** — [`encode_deltas`] / [`decode_deltas`] store a sorted
+//!   sequence (for example ledger numbers) as varint-encoded gaps.
+//! - **Deduplication** — [`dedupe_latest`] keeps one entry per key.
+//! - **Cleanup** — [`cleanup_expired`] drops entries older than a TTL, and
+//!   [`optimize`] runs cleanup plus deduplication in one pass.
+//! - **Monitoring & metrics** — [`StorageMetrics`] records what was saved.
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum StorageError {
+    /// The output buffer is too small.
+    BufferTooSmall,
+    /// The input sequence is not sorted ascending.
+    NotSorted,
+    /// The encoded input is truncated or malformed.
+    InvalidEncoding,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct Entry {
+    pub key: u64,
+    pub created_at: u64,
+    /// Stored size in bytes.
+    pub size: u32,
+}
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct StorageProfile {
+    pub entries: u32,
+    pub total_bytes: u64,
+    pub largest: u32,
+    pub distinct_keys: u32,
+}
+
+/// Analyze storage usage patterns.
+pub fn analyze(entries: &[Entry]) -> StorageProfile {
+    let mut profile = StorageProfile::default();
+    for (i, e) in entries.iter().enumerate() {
+        profile.entries += 1;
+        profile.total_bytes += e.size as u64;
+        profile.largest = profile.largest.max(e.size);
+        if !entries[..i].iter().any(|p| p.key == e.key) {
+            profile.distinct_keys += 1;
+        }
+    }
+    profile
+}
+
+fn write_varint(mut value: u32, out: &mut [u8], pos: &mut usize) -> Result<(), StorageError> {
+    loop {
+        if *pos >= out.len() {
+            return Err(StorageError::BufferTooSmall);
+        }
+        let byte = (value & 0x7f) as u8;
+        value >>= 7;
+        if value == 0 {
+            out[*pos] = byte;
+            *pos += 1;
+            return Ok(());
+        }
+        out[*pos] = byte | 0x80;
+        *pos += 1;
+    }
+}
+
+fn read_varint(input: &[u8], pos: &mut usize) -> Result<u32, StorageError> {
+    let mut value: u32 = 0;
+    let mut shift = 0;
+    loop {
+        let byte = *input.get(*pos).ok_or(StorageError::InvalidEncoding)?;
+        *pos += 1;
+        if shift >= 32 || (shift == 28 && byte & 0x70 != 0) {
+            return Err(StorageError::InvalidEncoding);
+        }
+        value |= ((byte & 0x7f) as u32) << shift;
+        if byte & 0x80 == 0 {
+            return Ok(value);
+        }
+        shift += 7;
+    }
+}
+
+/// Encode an ascending sequence as `[count][first][gap...]` varints.
+/// Returns the number of bytes written.
+pub fn encode_deltas(sorted: &[u32], out: &mut [u8]) -> Result<usize, StorageError> {
+    let mut pos = 0;
+    write_varint(sorted.len() as u32, out, &mut pos)?;
+    let mut previous = 0u32;
+    for (i, &value) in sorted.iter().enumerate() {
+        if i > 0 && value < previous {
+            return Err(StorageError::NotSorted);
+        }
+        write_varint(value - previous, out, &mut pos)?;
+        previous = value;
+    }
+    Ok(pos)
+}
+
+/// Decode a sequence produced by [`encode_deltas`]. Returns the number of values written.
+pub fn decode_deltas(input: &[u8], out: &mut [u32]) -> Result<usize, StorageError> {
+    let mut pos = 0;
+    let count = read_varint(input, &mut pos)? as usize;
+    if out.len() < count {
+        return Err(StorageError::BufferTooSmall);
+    }
+    let mut previous = 0u32;
+    for slot in out.iter_mut().take(count) {
+        let gap = read_varint(input, &mut pos)?;
+        previous = previous.checked_add(gap).ok_or(StorageError::InvalidEncoding)?;
+        *slot = previous;
+    }
+    if pos != input.len() {
+        return Err(StorageError::InvalidEncoding);
+    }
+    Ok(count)
+}
+
+/// Keep only the newest entry per key, preserving order of the survivors.
+/// Returns the new length; entries past it are unspecified.
+pub fn dedupe_latest(entries: &mut [Entry]) -> usize {
+    let mut kept = 0;
+    for i in 0..entries.len() {
+        let candidate = entries[i];
+        let found = entries[..kept].iter().position(|e| e.key == candidate.key);
+        match found {
+            Some(existing) => {
+                if candidate.created_at > entries[existing].created_at {
+                    entries[existing] = candidate;
+                }
+            }
+            None => {
+                entries[kept] = candidate;
+                kept += 1;
+            }
+        }
+    }
+    kept
+}
+
+/// Drop entries whose age is at least `ttl`. A `ttl` of zero keeps everything.
+/// Returns the new length; entries past it are unspecified.
+pub fn cleanup_expired(entries: &mut [Entry], now: u64, ttl: u64) -> usize {
+    if ttl == 0 {
+        return entries.len();
+    }
+    let mut kept = 0;
+    for i in 0..entries.len() {
+        let e = entries[i];
+        if now.saturating_sub(e.created_at) < ttl {
+            entries[kept] = e;
+            kept += 1;
+        }
+    }
+    kept
+}
+
+/// Cumulative storage savings.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct StorageMetrics {
+    pub runs: u32,
+    pub bytes_before: u64,
+    pub bytes_after: u64,
+    pub expired_removed: u32,
+    pub duplicates_removed: u32,
+}
+
+impl StorageMetrics {
+    /// Share of bytes saved, in basis points.
+    pub fn savings_bps(&self) -> u32 {
+        if self.bytes_before == 0 || self.bytes_after >= self.bytes_before {
+            return 0;
+        }
+        ((self.bytes_before - self.bytes_after) * 10_000 / self.bytes_before) as u32
+    }
+}
+
+/// Run cleanup then deduplication, record the result and return the new length.
+pub fn optimize(entries: &mut [Entry], now: u64, ttl: u64, metrics: &mut StorageMetrics) -> usize {
+    let before = analyze(entries);
+    let live = cleanup_expired(entries, now, ttl);
+    let unique = dedupe_latest(&mut entries[..live]);
+    let after = analyze(&entries[..unique]);
+
+    metrics.runs += 1;
+    metrics.bytes_before += before.total_bytes;
+    metrics.bytes_after += after.total_bytes;
+    metrics.expired_removed += (entries.len() - live) as u32;
+    metrics.duplicates_removed += (live - unique) as u32;
+    unique
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn entry(key: u64, created_at: u64, size: u32) -> Entry {
+        Entry { key, created_at, size }
+    }
+
+    #[test]
+    fn analyze_summarizes_usage() {
+        let entries = [entry(1, 0, 10), entry(2, 0, 30), entry(1, 5, 20)];
+        assert_eq!(
+            analyze(&entries),
+            StorageProfile { entries: 3, total_bytes: 60, largest: 30, distinct_keys: 2 }
+        );
+        assert_eq!(analyze(&[]), StorageProfile::default());
+    }
+
+    #[test]
+    fn delta_encoding_roundtrips_and_is_smaller() {
+        let ledgers: [u32; 8] = [1_000_000, 1_000_005, 1_000_006, 1_000_020, 1_000_021, 1_000_100, 1_000_101, 1_000_300];
+        let mut buf = [0u8; 64];
+        let n = encode_deltas(&ledgers, &mut buf).unwrap();
+        assert!(n < ledgers.len() * 4, "encoded {n} bytes");
+
+        let mut out = [0u32; 8];
+        assert_eq!(decode_deltas(&buf[..n], &mut out), Ok(8));
+        assert_eq!(out, ledgers);
+    }
+
+    #[test]
+    fn delta_encoding_handles_empty_and_extremes() {
+        let mut buf = [0u8; 16];
+        let n = encode_deltas(&[], &mut buf).unwrap();
+        let mut out = [0u32; 0];
+        assert_eq!(decode_deltas(&buf[..n], &mut out), Ok(0));
+
+        let extremes = [0u32, u32::MAX];
+        let n = encode_deltas(&extremes, &mut buf).unwrap();
+        let mut out = [0u32; 2];
+        assert_eq!(decode_deltas(&buf[..n], &mut out), Ok(2));
+        assert_eq!(out, extremes);
+    }
+
+    #[test]
+    fn delta_encoding_reports_errors() {
+        let mut buf = [0u8; 16];
+        assert_eq!(encode_deltas(&[5, 3], &mut buf), Err(StorageError::NotSorted));
+        let mut tiny = [0u8; 1];
+        assert_eq!(encode_deltas(&[1, 2], &mut tiny), Err(StorageError::BufferTooSmall));
+
+        let n = encode_deltas(&[1, 2, 3], &mut buf).unwrap();
+        let mut small_out = [0u32; 2];
+        assert_eq!(decode_deltas(&buf[..n], &mut small_out), Err(StorageError::BufferTooSmall));
+        let mut out = [0u32; 3];
+        assert_eq!(decode_deltas(&buf[..n - 1], &mut out), Err(StorageError::InvalidEncoding));
+        assert_eq!(decode_deltas(&[], &mut out), Err(StorageError::InvalidEncoding));
+        // Trailing garbage is rejected.
+        let mut padded = [0u8; 8];
+        padded[..n].copy_from_slice(&buf[..n]);
+        assert_eq!(decode_deltas(&padded[..n + 1], &mut out), Err(StorageError::InvalidEncoding));
+    }
+
+    #[test]
+    fn dedupe_keeps_newest_per_key() {
+        let mut entries = [entry(1, 10, 5), entry(2, 10, 5), entry(1, 20, 7), entry(3, 5, 5), entry(2, 1, 9)];
+        let n = dedupe_latest(&mut entries);
+        assert_eq!(n, 3);
+        assert_eq!(&entries[..n], &[entry(1, 20, 7), entry(2, 10, 5), entry(3, 5, 5)]);
+    }
+
+    #[test]
+    fn cleanup_removes_only_expired_entries() {
+        let mut entries = [entry(1, 0, 5), entry(2, 90, 5), entry(3, 40, 5)];
+        let n = cleanup_expired(&mut entries, 100, 50);
+        assert_eq!(n, 1);
+        assert_eq!(entries[0], entry(2, 90, 5));
+
+        let mut keep_all = [entry(1, 0, 5)];
+        assert_eq!(cleanup_expired(&mut keep_all, 1_000, 0), 1);
+    }
+
+    #[test]
+    fn optimize_runs_cleanup_and_dedup_and_records_metrics() {
+        let mut entries = [
+            entry(1, 0, 100),   // expired
+            entry(2, 90, 100),
+            entry(2, 95, 100),  // duplicate of key 2
+            entry(3, 99, 100),
+        ];
+        let mut metrics = StorageMetrics::default();
+        let n = optimize(&mut entries, 100, 50, &mut metrics);
+
+        assert_eq!(n, 2);
+        assert_eq!(metrics.runs, 1);
+        assert_eq!(metrics.expired_removed, 1);
+        assert_eq!(metrics.duplicates_removed, 1);
+        assert_eq!(metrics.bytes_before, 400);
+        assert_eq!(metrics.bytes_after, 200);
+        assert_eq!(metrics.savings_bps(), 5_000);
+        assert_eq!(StorageMetrics::default().savings_bps(), 0);
+    }
+}