diff --git a/contracts/split/src/aggregation.rs b/contracts/split/src/aggregation.rs
new file mode 100644
index 0000000..e72ec7d
--- /dev/null
+++ b/contracts/split/src/aggregation.rs
@@ -0,0 +1,313 @@
+//! Request aggregation (issue #1022).
+//!
+//! Collects amount-bearing requests into a bounded batch and coalesces
+//! requests that target the same destination so a single downstream
+//! operation (e.g. one token transfer) can serve many of them.
+//!
+//! - **Batching** — [`Aggregator::add`] queues up to [`MAX_BATCH`] requests.
+//! - **Timeouts** — a batch is due once it is full or `window_ledgers` have
+//!   passed since the first request ([`Aggregator::is_due`]); adding to a
+//!   timed-out batch is refused until it is flushed.
+//! - **Aggregation logic** — [`Aggregator::flush`] groups by target and sums
+//!   amounts with overflow checks.
+//! - **Result distribution** — [`Flush::group_of`] maps each request back to
+//!   the group whose outcome applies to it.
+//! - **Metrics** — [`AggregationMetrics`] reports how many operations the
+//!   batching saved.
+
+/// Maximum number of requests per batch.
+pub const MAX_BATCH: usize = 32;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AggregationError {
+    /// `window_ledgers` or `max_size` is zero, or `max_size` exceeds [`MAX_BATCH`].
+    InvalidConfig,
+    /// Amount is zero or negative.
+    InvalidAmount,
+    /// The batch is full or has timed out and must be flushed first.
+    BatchClosed,
+    /// Summing amounts overflowed `i128`.
+    Overflow,
+    /// There is nothing to flush.
+    Empty,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct Request {
+    pub id: u32,
+    pub target: u32,
+    pub amount: i128,
+}
+
+impl Request {
+    const EMPTY: Request = Request { id: 0, target: 0, amount: 0 };
+}
+
+/// Combined amount for one target.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct Group {
+    pub target: u32,
+    pub total: i128,
+    pub requests: u32,
+}
+
+impl Group {
+    const EMPTY: Group = Group { target: 0, total: 0, requests: 0 };
+}
+
+/// Result of flushing a batch.
+#[derive(Clone, Copy, Debug)]
+pub struct Flush {
+    groups: [Group; MAX_BATCH],
+    group_count: usize,
+    request_group: [u8; MAX_BATCH],
+    request_count: usize,
+    /// `true` when the flush was triggered by the window elapsing.
+    pub timed_out: bool,
+}
+
+impl Flush {
+    pub fn groups(&self) -> &[Group] {
+        &self.groups[..self.group_count]
+    }
+
+    /// Index into [`Flush::groups`] whose outcome applies to request `index`
+    /// (in the order requests were added).
+    pub fn group_of(&self, index: usize) -> Option<usize> {
+        if index < self.request_count {
+            Some(self.request_group[index] as usize)
+        } else {
+            None
+        }
+    }
+
+    pub fn request_count(&self) -> usize {
+        self.request_count
+    }
+}
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct AggregationMetrics {
+    pub requests: u32,
+    pub batches: u32,
+    pub groups: u32,
+    pub timeouts: u32,
+    pub rejected: u32,
+}
+
+impl AggregationMetrics {
+    /// Downstream operations avoided by aggregating (`requests - groups`).
+    pub fn operations_saved(&self) -> u32 {
+        self.requests.saturating_sub(self.groups)
+    }
+}
+
+pub struct Aggregator {
+    window_ledgers: u32,
+    max_size: usize,
+    opened_at: u32,
+    requests: [Request; MAX_BATCH],
+    len: usize,
+    metrics: AggregationMetrics,
+}
+
+impl Aggregator {
+    pub fn new(window_ledgers: u32, max_size: usize) -> Result<Self, AggregationError> {
+        if window_ledgers == 0 || max_size == 0 || max_size > MAX_BATCH {
+            return Err(AggregationError::InvalidConfig);
+        }
+        Ok(Aggregator {
+            window_ledgers,
+            max_size,
+            opened_at: 0,
+            requests: [Request::EMPTY; MAX_BATCH],
+            len: 0,
+            metrics: AggregationMetrics::default(),
+        })
+    }
+
+    pub fn metrics(&self) -> AggregationMetrics {
+        self.metrics
+    }
+
+    pub fn pending(&self) -> usize {
+        self.len
+    }
+
+    /// `true` when the batch is full or its window has elapsed.
+    pub fn is_due(&self, now: u32) -> bool {
+        self.len > 0 && (self.len >= self.max_size || self.window_elapsed(now))
+    }
+
+    fn window_elapsed(&self, now: u32) -> bool {
+        self.len > 0 && now.saturating_sub(self.opened_at) >= self.window_ledgers
+    }
+
+    /// Queue a request. Returns the number of pending requests.
+    pub fn add(&mut self, request: Request, now: u32) -> Result<usize, AggregationError> {
+        if request.amount <= 0 {
+            self.metrics.rejected += 1;
+            return Err(AggregationError::InvalidAmount);
+        }
+        if self.len >= self.max_size || self.window_elapsed(now) {
+            self.metrics.rejected += 1;
+            return Err(AggregationError::BatchClosed);
+        }
+        if self.len == 0 {
+            self.opened_at = now;
+        }
+        self.requests[self.len] = request;
+        self.len += 1;
+        self.metrics.requests += 1;
+        Ok(self.len)
+    }
+
+    /// Coalesce pending requests by target and reset the batch.
+    pub fn flush(&mut self, now: u32) -> Result<Flush, AggregationError> {
+        if self.len == 0 {
+            return Err(AggregationError::Empty);
+        }
+        let timed_out = self.window_elapsed(now) && self.len < self.max_size;
+
+        let mut groups = [Group::EMPTY; MAX_BATCH];
+        let mut group_count = 0usize;
+        let mut request_group = [0u8; MAX_BATCH];
+
+        for i in 0..self.len {
+            let req = self.requests[i];
+            let mut found = None;
+            for (g, group) in groups.iter().enumerate().take(group_count) {
+                if group.target == req.target {
+                    found = Some(g);
+                    break;
+                }
+            }
+            let g = match found {
+                Some(g) => g,
+                None => {
+                    groups[group_count] = Group { target: req.target, total: 0, requests: 0 };
+                    group_count += 1;
+                    group_count - 1
+                }
+            };
+            groups[g].total = groups[g]
+                .total
+                .checked_add(req.amount)
+                .ok_or(AggregationError::Overflow)?;
+            groups[g].requests += 1;
+            request_group[i] = g as u8;
+        }
+
+        let request_count = self.len;
+        self.len = 0;
+        self.metrics.batches += 1;
+        self.metrics.groups += group_count as u32;
+        if timed_out {
+            self.metrics.timeouts += 1;
+        }
+
+        Ok(Flush { groups, group_count, request_group, request_count, timed_out })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn req(id: u32, target: u32, amount: i128) -> Request {
+        Request { id, target, amount }
+    }
+
+    #[test]
+    fn rejects_invalid_config_and_amounts() {
+        assert_eq!(Aggregator::new(0, 4).err(), Some(AggregationError::InvalidConfig));
+        assert_eq!(Aggregator::new(10, 0).err(), Some(AggregationError::InvalidConfig));
+        assert_eq!(Aggregator::new(10, MAX_BATCH + 1).err(), Some(AggregationError::InvalidConfig));
+
+        let mut a = Aggregator::new(10, 4).unwrap();
+        assert_eq!(a.add(req(1, 1, 0), 0), Err(AggregationError::InvalidAmount));
+        assert_eq!(a.add(req(1, 1, -5), 0), Err(AggregationError::InvalidAmount));
+        assert_eq!(a.metrics().rejected, 2);
+    }
+
+    #[test]
+    fn coalesces_same_target_and_distributes_results() {
+        let mut a = Aggregator::new(10, 8).unwrap();
+        a.add(req(1, 100, 10), 0).unwrap();
+        a.add(req(2, 200, 5), 1).unwrap();
+        a.add(req(3, 100, 7), 2).unwrap();
+        assert_eq!(a.pending(), 3);
+
+        let flush = a.flush(3).unwrap();
+        assert_eq!(flush.groups().len(), 2);
+        assert_eq!(flush.groups()[0], Group { target: 100, total: 17, requests: 2 });
+        assert_eq!(flush.groups()[1], Group { target: 200, total: 5, requests: 1 });
+        assert_eq!(flush.group_of(0), Some(0));
+        assert_eq!(flush.group_of(1), Some(1));
+        assert_eq!(flush.group_of(2), Some(0));
+        assert_eq!(flush.group_of(3), None);
+        assert!(!flush.timed_out);
+        assert_eq!(a.pending(), 0);
+        assert_eq!(a.metrics().operations_saved(), 1);
+    }
+
+    #[test]
+    fn full_batch_is_due_and_refuses_more() {
+        let mut a = Aggregator::new(100, 2).unwrap();
+        a.add(req(1, 1, 1), 0).unwrap();
+        assert!(!a.is_due(0));
+        a.add(req(2, 2, 1), 0).unwrap();
+        assert!(a.is_due(0));
+        assert_eq!(a.add(req(3, 3, 1), 0), Err(AggregationError::BatchClosed));
+        a.flush(0).unwrap();
+        assert_eq!(a.add(req(3, 3, 1), 0), Ok(1));
+    }
+
+    #[test]
+    fn window_timeout_is_respected() {
+        let mut a = Aggregator::new(5, 8).unwrap();
+        a.add(req(1, 1, 1), 10).unwrap();
+        assert!(!a.is_due(14));
+        assert!(a.is_due(15));
+        assert_eq!(a.add(req(2, 1, 1), 15), Err(AggregationError::BatchClosed));
+
+        let flush = a.flush(15).unwrap();
+        assert!(flush.timed_out);
+        assert_eq!(a.metrics().timeouts, 1);
+
+        // A new batch opens a fresh window.
+        a.add(req(3, 1, 1), 20).unwrap();
+        assert!(!a.is_due(24));
+    }
+
+    #[test]
+    fn empty_flush_is_an_error() {
+        let mut a = Aggregator::new(5, 8).unwrap();
+        assert_eq!(a.flush(0).err(), Some(AggregationError::Empty));
+        assert!(!a.is_due(1_000));
+    }
+
+    #[test]
+    fn overflow_is_reported_without_wrapping() {
+        let mut a = Aggregator::new(5, 8).unwrap();
+        a.add(req(1, 1, i128::MAX), 0).unwrap();
+        a.add(req(2, 1, 1), 0).unwrap();
+        assert_eq!(a.flush(0).err(), Some(AggregationError::Overflow));
+    }
+
+    #[test]
+    fn metrics_accumulate_across_batches() {
+        let mut a = Aggregator::new(5, 8).unwrap();
+        for batch in 0..3u32 {
+            for i in 0..4 {
+                a.add(req(i, i % 2, 1), batch * 10).unwrap();
+            }
+            a.flush(batch * 10).unwrap();
+        }
+        let m = a.metrics();
+        assert_eq!(m.requests, 12);
+        assert_eq!(m.batches, 3);
+        assert_eq!(m.groups, 6);
+        assert_eq!(m.operations_saved(), 6);
+    }
+}
diff --git a/contracts/split/src/compression.rs b/contracts/split/src/compression.rs
new file mode 100644
index 0000000..882cbf0
--- /dev/null
+++ b/contracts/split/src/compression.rs
@@ -0,0 +1,251 @@
+//! Payload compression (issue #1021).
+//!
+//! Run-length encoding for byte payloads with explicit negotiation, raw
+//! fallback and metrics. Heap free and `no_std` friendly: callers pass the
+//! output buffer.
+//!
+//! Wire format: `[encoding byte][body...]`
+//! - `0x00` raw: body is the original bytes.
+//! - `0x01` RLE: body is `(run_length, value)` pairs, run length 1..=255.
+//!
+//! Negotiation: the caller advertises supported encodings as a bitmask
+//! ([`ACCEPT_RAW`], [`ACCEPT_RLE`]). RLE is used only when it is accepted and
+//! strictly smaller than raw; otherwise [`compress`] falls back to raw.
+
+pub const ENCODING_RAW: u8 = 0;
+pub const ENCODING_RLE: u8 = 1;
+
+pub const ACCEPT_RAW: u8 = 0b01;
+pub const ACCEPT_RLE: u8 = 0b10;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum CompressionError {
+    /// The output buffer is too small.
+    BufferTooSmall,
+    /// The payload is empty, truncated or uses an unknown encoding.
+    InvalidPayload,
+    /// The accepted-encodings mask has no supported encoding.
+    NoAcceptableEncoding,
+}
+
+/// Pick the best encoding allowed by `accepted` for `input`.
+pub fn negotiate(accepted: u8, input: &[u8]) -> Result<u8, CompressionError> {
+    if accepted & ACCEPT_RLE != 0 && rle_size(input) < input.len() {
+        Ok(ENCODING_RLE)
+    } else if accepted & ACCEPT_RAW != 0 {
+        Ok(ENCODING_RAW)
+    } else if accepted & ACCEPT_RLE != 0 {
+        // Only RLE is acceptable even though it is not smaller.
+        Ok(ENCODING_RLE)
+    } else {
+        Err(CompressionError::NoAcceptableEncoding)
+    }
+}
+
+fn rle_size(input: &[u8]) -> usize {
+    let mut size = 0;
+    let mut i = 0;
+    while i < input.len() {
+        let mut run = 1;
+        while i + run < input.len() && input[i + run] == input[i] && run < 255 {
+            run += 1;
+        }
+        size += 2;
+        i += run;
+    }
+    size
+}
+
+/// Compress `input` into `out`, returning the number of bytes written.
+pub fn compress(accepted: u8, input: &[u8], out: &mut [u8]) -> Result<usize, CompressionError> {
+    let encoding = negotiate(accepted, input)?;
+    if out.is_empty() {
+        return Err(CompressionError::BufferTooSmall);
+    }
+    out[0] = encoding;
+
+    if encoding == ENCODING_RAW {
+        if out.len() < 1 + input.len() {
+            return Err(CompressionError::BufferTooSmall);
+        }
+        out[1..1 + input.len()].copy_from_slice(input);
+        return Ok(1 + input.len());
+    }
+
+    if out.len() < 1 + rle_size(input) {
+        return Err(CompressionError::BufferTooSmall);
+    }
+    let mut w = 1;
+    let mut i = 0;
+    while i < input.len() {
+        let mut run = 1;
+        while i + run < input.len() && input[i + run] == input[i] && run < 255 {
+            run += 1;
+        }
+        out[w] = run as u8;
+        out[w + 1] = input[i];
+        w += 2;
+        i += run;
+    }
+    Ok(w)
+}
+
+/// Decompress `input` into `out`, returning the number of bytes written.
+pub fn decompress(input: &[u8], out: &mut [u8]) -> Result<usize, CompressionError> {
+    let (&encoding, body) = input.split_first().ok_or(CompressionError::InvalidPayload)?;
+    match encoding {
+        ENCODING_RAW => {
+            if out.len() < body.len() {
+                return Err(CompressionError::BufferTooSmall);
+            }
+            out[..body.len()].copy_from_slice(body);
+            Ok(body.len())
+        }
+        ENCODING_RLE => {
+            if body.len() % 2 != 0 {
+                return Err(CompressionError::InvalidPayload);
+            }
+            let mut w = 0;
+            for pair in body.chunks(2) {
+                let run = pair[0] as usize;
+                if run == 0 {
+                    return Err(CompressionError::InvalidPayload);
+                }
+                if out.len() < w + run {
+                    return Err(CompressionError::BufferTooSmall);
+                }
+                for b in &mut out[w..w + run] {
+                    *b = pair[1];
+                }
+                w += run;
+            }
+            Ok(w)
+        }
+        _ => Err(CompressionError::InvalidPayload),
+    }
+}
+
+/// Cumulative compression counters.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct CompressionMetrics {
+    pub payloads: u32,
+    pub rle_payloads: u32,
+    pub raw_fallbacks: u32,
+    pub input_bytes: u64,
+    pub output_bytes: u64,
+}
+
+impl CompressionMetrics {
+    /// Compress and record the result.
+    pub fn compress(
+        &mut self,
+        accepted: u8,
+        input: &[u8],
+        out: &mut [u8],
+    ) -> Result<usize, CompressionError> {
+        let written = compress(accepted, input, out)?;
+        self.payloads += 1;
+        if out[0] == ENCODING_RLE {
+            self.rle_payloads += 1;
+        } else {
+            self.raw_fallbacks += 1;
+        }
+        self.input_bytes += input.len() as u64;
+        self.output_bytes += written as u64;
+        Ok(written)
+    }
+
+    /// Size reduction in basis points (10_000 = 100%). `0` when nothing was saved.
+    pub fn savings_bps(&self) -> u32 {
+        if self.input_bytes == 0 || self.output_bytes >= self.input_bytes {
+            return 0;
+        }
+        ((self.input_bytes - self.output_bytes) * 10_000 / self.input_bytes) as u32
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn roundtrip(accepted: u8, input: &[u8]) -> (usize, u8) {
+        let mut packed = [0u8; 600];
+        let n = compress(accepted, input, &mut packed).unwrap();
+        let mut restored = [0u8; 600];
+        let m = decompress(&packed[..n], &mut restored).unwrap();
+        assert_eq!(&restored[..m], input);
+        (n, packed[0])
+    }
+
+    #[test]
+    fn repetitive_data_shrinks_by_more_than_70_percent() {
+        let input = [7u8; 300];
+        let (n, enc) = roundtrip(ACCEPT_RAW | ACCEPT_RLE, &input);
+        assert_eq!(enc, ENCODING_RLE);
+        assert!(n * 100 <= input.len() * 30, "compressed {n} of {}", input.len());
+    }
+
+    #[test]
+    fn incompressible_data_falls_back_to_raw() {
+        let mut input = [0u8; 64];
+        for (i, b) in input.iter_mut().enumerate() {
+            *b = i as u8;
+        }
+        let (n, enc) = roundtrip(ACCEPT_RAW | ACCEPT_RLE, &input);
+        assert_eq!(enc, ENCODING_RAW);
+        assert_eq!(n, input.len() + 1);
+    }
+
+    #[test]
+    fn negotiation_respects_accepted_encodings() {
+        let repetitive = [1u8; 50];
+        assert_eq!(negotiate(ACCEPT_RAW, &repetitive), Ok(ENCODING_RAW));
+        assert_eq!(negotiate(ACCEPT_RLE, &repetitive), Ok(ENCODING_RLE));
+        assert_eq!(negotiate(ACCEPT_RAW | ACCEPT_RLE, &repetitive), Ok(ENCODING_RLE));
+        assert_eq!(negotiate(0, &repetitive), Err(CompressionError::NoAcceptableEncoding));
+        // RLE-only peers still get RLE even when it is larger.
+        assert_eq!(negotiate(ACCEPT_RLE, &[1, 2, 3]), Ok(ENCODING_RLE));
+    }
+
+    #[test]
+    fn runs_longer_than_255_are_split() {
+        roundtrip(ACCEPT_RLE, &[9u8; 520]);
+    }
+
+    #[test]
+    fn empty_input_roundtrips() {
+        let (n, _) = roundtrip(ACCEPT_RAW | ACCEPT_RLE, &[]);
+        assert_eq!(n, 1);
+    }
+
+    #[test]
+    fn malformed_payloads_are_rejected() {
+        let mut out = [0u8; 16];
+        assert_eq!(decompress(&[], &mut out), Err(CompressionError::InvalidPayload));
+        assert_eq!(decompress(&[9, 1], &mut out), Err(CompressionError::InvalidPayload));
+        assert_eq!(decompress(&[ENCODING_RLE, 3], &mut out), Err(CompressionError::InvalidPayload));
+        assert_eq!(decompress(&[ENCODING_RLE, 0, 5], &mut out), Err(CompressionError::InvalidPayload));
+        assert_eq!(decompress(&[ENCODING_RLE, 200, 5], &mut out), Err(CompressionError::BufferTooSmall));
+    }
+
+    #[test]
+    fn small_output_buffer_is_reported() {
+        let mut tiny = [0u8; 1];
+        assert_eq!(compress(ACCEPT_RAW, &[1, 2, 3], &mut tiny), Err(CompressionError::BufferTooSmall));
+    }
+
+    #[test]
+    fn metrics_track_savings_and_fallbacks() {
+        let mut m = CompressionMetrics::default();
+        let mut out = [0u8; 600];
+        m.compress(ACCEPT_RAW | ACCEPT_RLE, &[0u8; 200], &mut out).unwrap();
+        m.compress(ACCEPT_RAW | ACCEPT_RLE, &[1, 2, 3, 4], &mut out).unwrap();
+        assert_eq!(m.payloads, 2);
+        assert_eq!(m.rle_payloads, 1);
+        assert_eq!(m.raw_fallbacks, 1);
+        assert_eq!(m.input_bytes, 204);
+        assert_eq!(m.output_bytes, 3 + 5);
+        assert!(m.savings_bps() > 9_000);
+        assert_eq!(CompressionMetrics::default().savings_bps(), 0);
+    }
+}
diff --git a/contracts/split/src/error_scenario_tests.rs b/contracts/split/src/error_scenario_tests.rs
new file mode 100644
index 0000000..859fbca
--- /dev/null
+++ b/contracts/split/src/error_scenario_tests.rs
@@ -0,0 +1,249 @@
+#![cfg(test)]
+
+//! Error scenario testing (issue #1023).
+//!
+//! Every scenario injects one invalid operation against a known-good invoice
+//! and checks that:
+//! 1. the operation is rejected (**error injection**),
+//! 2. the invoice is left exactly as it was (**no partial state**), and
+//! 3. a valid operation still succeeds afterwards (**recovery**).
+//!
+//! Outcomes are collected in an [`ErrorReport`] (**reporting**) that is
+//! grouped by [`ErrorCategory`] (**analytics**). The scenario catalogue is
+//! documented in `docs/ISSUE_1023_ERROR_SCENARIOS.md`.
+
+extern crate std;
+
+use crate::test::default_options;
+use crate::types::InvoiceStatus;
+use crate::{SplitContract, SplitContractClient};
+use soroban_sdk::{
+    testutils::{Address as _, Ledger},
+    token::StellarAssetClient,
+    Address, Env, Vec,
+};
+
+const DEADLINE: u64 = 2_000;
+const AMOUNT: i128 = 100;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum ErrorCategory {
+    InvalidAmount,
+    InvalidInput,
+    #[allow(dead_code)]
+    InvalidState,
+    Timing,
+    NotFound,
+}
+
+struct Scenario {
+    name: &'static str,
+    category: ErrorCategory,
+    /// Injects the error; returns `true` when the contract rejected it.
+    inject: fn(&Fixture) -> bool,
+}
+
+#[derive(Default)]
+struct ErrorReport {
+    rejected: std::vec::Vec<&'static str>,
+    accepted: std::vec::Vec<&'static str>,
+    recovered: u32,
+    by_category: std::vec::Vec<(ErrorCategory, u32)>,
+}
+
+impl ErrorReport {
+    fn record(&mut self, s: &Scenario, rejected: bool, recovered: bool) {
+        if rejected {
+            self.rejected.push(s.name);
+        } else {
+            self.accepted.push(s.name);
+        }
+        if recovered {
+            self.recovered += 1;
+        }
+        match self.by_category.iter_mut().find(|(c, _)| *c == s.category) {
+            Some((_, n)) => *n += 1,
+            None => self.by_category.push((s.category, 1)),
+        }
+    }
+
+    fn count(&self, category: ErrorCategory) -> u32 {
+        self.by_category.iter().find(|(c, _)| *c == category).map(|(_, n)| *n).unwrap_or(0)
+    }
+}
+
+struct Fixture<'a> {
+    env: &'a Env,
+    c: SplitContractClient<'a>,
+    token: Address,
+    id: u64,
+    creator: Address,
+    payer: Address,
+}
+
+fn fixture(env: &Env) -> Fixture<'_> {
+    env.mock_all_auths();
+    let contract_id = env.register(SplitContract, ());
+    let token_admin = Address::generate(env);
+    let token = env.register_stellar_asset_contract_v2(token_admin.clone()).address();
+    StellarAssetClient::new(env, &token).mint(&token_admin, &1_000_000);
+
+    let c = SplitContractClient::new(env, &contract_id);
+    c.initialize(
+        &Address::generate(env),
+        &0_i128,
+        &Address::generate(env),
+        &token,
+        &0_u32,
+        &None,
+        &0_u32,
+        &0_u32,
+        &0_u64,
+    );
+    env.ledger().set_timestamp(1_000);
+
+    let creator = Address::generate(env);
+    let mut recipients = Vec::new(env);
+    recipients.push_back(Address::generate(env));
+    let mut amounts = Vec::new(env);
+    amounts.push_back(AMOUNT);
+    let id = c.create_invoice(&creator, &recipients, &amounts, &token, &DEADLINE, &default_options(env));
+
+    let payer = Address::generate(env);
+    StellarAssetClient::new(env, &token).mint(&payer, &10_000);
+
+    Fixture { env, c, token, id, creator, payer }
+}
+
+fn pay(c: &SplitContractClient, payer: &Address, id: u64, amount: i128) -> bool {
+    c.try_pay(payer, &id, &amount, &0_u64, &false, &false, &None).is_ok()
+}
+
+fn scenarios() -> [Scenario; 8] {
+    [
+        Scenario {
+            name: "pay zero",
+            category: ErrorCategory::InvalidAmount,
+            inject: |f| !pay(&f.c, &f.payer, f.id, 0),
+        },
+        Scenario {
+            name: "pay negative",
+            category: ErrorCategory::InvalidAmount,
+            inject: |f| !pay(&f.c, &f.payer, f.id, -1),
+        },
+        Scenario {
+            name: "pay more than remaining",
+            category: ErrorCategory::InvalidAmount,
+            inject: |f| !pay(&f.c, &f.payer, f.id, AMOUNT + 1),
+        },
+        Scenario {
+            name: "pay after deadline",
+            category: ErrorCategory::Timing,
+            inject: |f| {
+                f.env.ledger().set_timestamp(DEADLINE + 1);
+                let rejected = !pay(&f.c, &f.payer, f.id, 10);
+                f.env.ledger().set_timestamp(1_000);
+                rejected
+            },
+        },
+        Scenario {
+            name: "refund before deadline",
+            category: ErrorCategory::Timing,
+            inject: |f| f.c.try_refund(&f.id).is_err(),
+        },
+        Scenario {
+            name: "create invoice with zero amount",
+            category: ErrorCategory::InvalidInput,
+            inject: |f| {
+                let mut recipients = Vec::new(f.env);
+                recipients.push_back(Address::generate(f.env));
+                let mut amounts = Vec::new(f.env);
+                amounts.push_back(0_i128);
+                f.c.try_create_invoice(&f.creator, &recipients, &amounts, &f.token, &DEADLINE, &default_options(f.env))
+                    .is_err()
+            },
+        },
+        Scenario {
+            name: "create invoice with mismatched recipients and amounts",
+            category: ErrorCategory::InvalidInput,
+            inject: |f| {
+                let mut recipients = Vec::new(f.env);
+                recipients.push_back(Address::generate(f.env));
+                let amounts: Vec<i128> = Vec::new(f.env);
+                f.c.try_create_invoice(&f.creator, &recipients, &amounts, &f.token, &DEADLINE, &default_options(f.env))
+                    .is_err()
+            },
+        },
+        Scenario {
+            name: "pay an unknown invoice",
+            category: ErrorCategory::NotFound,
+            inject: |f| !pay(&f.c, &f.payer, f.id + 1_000, 10),
+        },
+    ]
+}
+
+#[test]
+fn injected_errors_are_rejected_leave_state_intact_and_recover() {
+    let mut report = ErrorReport::default();
+
+    for scenario in scenarios().iter() {
+        let env = Env::default();
+        let f = fixture(&env);
+        let before = f.c.get_invoice(&f.id);
+
+        let rejected = (scenario.inject)(&f);
+
+        let after = f.c.get_invoice(&f.id);
+        assert_eq!(after.funded, before.funded, "{}: funded changed", scenario.name);
+        assert_eq!(after.status, before.status, "{}: status changed", scenario.name);
+
+        // Recovery: a valid payment still completes the invoice.
+        let recovered = pay(&f.c, &f.payer, f.id, AMOUNT)
+            && f.c.get_invoice(&f.id).status == InvoiceStatus::Released;
+
+        report.record(scenario, rejected, recovered);
+    }
+
+    assert!(report.accepted.is_empty(), "errors that were not rejected: {:?}", report.accepted);
+    assert_eq!(report.rejected.len(), scenarios().len());
+    assert_eq!(report.recovered as usize, scenarios().len());
+}
+
+#[test]
+fn error_analytics_group_scenarios_by_category() {
+    let mut report = ErrorReport::default();
+    for scenario in scenarios().iter() {
+        report.record(scenario, true, true);
+    }
+
+    assert_eq!(report.count(ErrorCategory::InvalidAmount), 3);
+    assert_eq!(report.count(ErrorCategory::Timing), 2);
+    assert_eq!(report.count(ErrorCategory::InvalidState), 0);
+    assert_eq!(report.count(ErrorCategory::InvalidInput), 2);
+    assert_eq!(report.count(ErrorCategory::NotFound), 1);
+    let total: u32 = report.by_category.iter().map(|(_, n)| *n).sum();
+    assert_eq!(total as usize, scenarios().len());
+}
+
+#[test]
+fn cancel_after_full_payment_is_rejected_and_state_is_kept() {
+    let env = Env::default();
+    let f = fixture(&env);
+    assert!(pay(&f.c, &f.payer, f.id, AMOUNT));
+    assert_eq!(f.c.get_invoice(&f.id).status, InvoiceStatus::Released);
+
+    assert!(f.c.try_cancel_invoice(&f.creator, &f.id).is_err());
+    assert_eq!(f.c.get_invoice(&f.id).status, InvoiceStatus::Released);
+}
+
+#[test]
+fn double_refund_is_rejected() {
+    let env = Env::default();
+    let f = fixture(&env);
+    assert!(pay(&f.c, &f.payer, f.id, 40));
+
+    env.ledger().set_timestamp(DEADLINE + 1);
+    assert!(f.c.try_refund(&f.id).is_ok());
+    assert!(f.c.try_refund(&f.id).is_err());
+    assert_eq!(f.c.get_invoice(&f.id).status, InvoiceStatus::Refunded);
+}
diff --git a/contracts/split/src/lib.rs b/contracts/split/src/lib.rs
index b2b16d0..f176d51 100644
--- a/contracts/split/src/lib.rs
+++ b/contracts/split/src/lib.rs
@@ -83,6 +83,9 @@ pub mod security_audit;
 pub mod payment_tracking;
 pub mod velocity;
 pub mod split_rules;
+pub mod pool;
+pub mod compression;
+pub mod aggregation;
 
 #[cfg(test)]
 mod test;
diff --git a/contracts/split/src/pool.rs b/contracts/split/src/pool.rs
new file mode 100644
index 0000000..d8fff01
--- /dev/null
+++ b/contracts/split/src/pool.rs
@@ -0,0 +1,321 @@
+//! Resource pooling (issue #1020).
+//!
+//! A bounded pool of reusable execution slots, e.g. for batch operations that
+//! need a limited number of concurrent workers. The pool is fixed-size, heap
+//! free and `no_std` friendly, and time is expressed in ledger sequence numbers.
+//!
+//! - **Initialization** — [`Pool::new`] validates the [`PoolConfig`] and
+//!   creates `capacity` primary slots followed by `standby` failover slots.
+//! - **Recycling** — idle slots older than `max_age` ledgers, or used
+//!   `max_uses` times, are reset by [`Pool::recycle`] (also run on acquire).
+//! - **Health checks** — a slot that fails `failure_threshold` times in a row
+//!   becomes unhealthy; [`Pool::health_check`] reports and repairs idle ones.
+//! - **Failover** — [`Pool::acquire`] falls back to standby slots when every
+//!   healthy primary slot is busy or unhealthy.
+//! - **Monitoring** — [`Pool::stats`] exposes cumulative counters.
+
+/// Hard upper bound on the number of slots (primary + standby).
+pub const MAX_POOL_SLOTS: usize = 16;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum PoolError {
+    /// `capacity` is zero or `capacity + standby` exceeds [`MAX_POOL_SLOTS`].
+    InvalidConfig,
+    /// Every usable slot is busy or unhealthy.
+    Exhausted,
+    /// The slot index is out of range.
+    InvalidSlot,
+    /// The slot is not currently in use.
+    NotInUse,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct PoolConfig {
+    /// Number of primary slots.
+    pub capacity: usize,
+    /// Number of standby slots used only for failover.
+    pub standby: usize,
+    /// Idle slots older than this many ledgers are recycled. `0` disables.
+    pub max_age: u32,
+    /// Slots used this many times are recycled. `0` disables.
+    pub max_uses: u32,
+    /// Consecutive failures before a slot is marked unhealthy. `0` disables.
+    pub failure_threshold: u32,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum SlotState {
+    Idle,
+    InUse,
+    Unhealthy,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct Slot {
+    pub state: SlotState,
+    pub created_at: u32,
+    pub uses: u32,
+    pub failures: u32,
+    pub standby: bool,
+}
+
+impl Slot {
+    fn fresh(now: u32, standby: bool) -> Self {
+        Slot { state: SlotState::Idle, created_at: now, uses: 0, failures: 0, standby }
+    }
+}
+
+/// Cumulative pool counters for monitoring.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct PoolStats {
+    pub acquired: u32,
+    pub released: u32,
+    pub recycled: u32,
+    pub failovers: u32,
+    pub exhausted: u32,
+    pub marked_unhealthy: u32,
+}
+
+/// Point-in-time view of slot states.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct HealthReport {
+    pub idle: u32,
+    pub in_use: u32,
+    pub unhealthy: u32,
+    /// `true` when at least one healthy primary slot exists.
+    pub healthy: bool,
+}
+
+pub struct Pool {
+    config: PoolConfig,
+    slots: [Slot; MAX_POOL_SLOTS],
+    len: usize,
+    stats: PoolStats,
+}
+
+impl Pool {
+    pub fn new(config: PoolConfig, now: u32) -> Result<Self, PoolError> {
+        let len = config.capacity + config.standby;
+        if config.capacity == 0 || len > MAX_POOL_SLOTS {
+            return Err(PoolError::InvalidConfig);
+        }
+        let mut slots = [Slot::fresh(now, false); MAX_POOL_SLOTS];
+        for (i, slot) in slots.iter_mut().enumerate().take(len) {
+            *slot = Slot::fresh(now, i >= config.capacity);
+        }
+        Ok(Pool { config, slots, len, stats: PoolStats::default() })
+    }
+
+    pub fn stats(&self) -> PoolStats {
+        self.stats
+    }
+
+    pub fn slot(&self, index: usize) -> Option<Slot> {
+        if index < self.len {
+            Some(self.slots[index])
+        } else {
+            None
+        }
+    }
+
+    fn needs_recycle(&self, slot: &Slot, now: u32) -> bool {
+        let aged = self.config.max_age > 0 && now.saturating_sub(slot.created_at) >= self.config.max_age;
+        let worn = self.config.max_uses > 0 && slot.uses >= self.config.max_uses;
+        aged || worn
+    }
+
+    /// Reset idle slots that are too old or too used. Returns how many were recycled.
+    pub fn recycle(&mut self, now: u32) -> u32 {
+        let mut recycled = 0;
+        for i in 0..self.len {
+            let slot = self.slots[i];
+            if slot.state == SlotState::Idle && self.needs_recycle(&slot, now) {
+                self.slots[i] = Slot::fresh(now, slot.standby);
+                recycled += 1;
+            }
+        }
+        self.stats.recycled += recycled;
+        recycled
+    }
+
+    /// Acquire a slot, preferring healthy primary slots and failing over to standby.
+    pub fn acquire(&mut self, now: u32) -> Result<usize, PoolError> {
+        self.recycle(now);
+
+        for pass_standby in [false, true] {
+            for i in 0..self.len {
+                let slot = &mut self.slots[i];
+                if slot.standby == pass_standby && slot.state == SlotState::Idle {
+                    slot.state = SlotState::InUse;
+                    self.stats.acquired += 1;
+                    if pass_standby {
+                        self.stats.failovers += 1;
+                    }
+                    return Ok(i);
+                }
+            }
+        }
+
+        self.stats.exhausted += 1;
+        Err(PoolError::Exhausted)
+    }
+
+    /// Return a slot to the pool. `ok == false` records a failure against it.
+    pub fn release(&mut self, index: usize, ok: bool) -> Result<(), PoolError> {
+        if index >= self.len {
+            return Err(PoolError::InvalidSlot);
+        }
+        let threshold = self.config.failure_threshold;
+        let slot = &mut self.slots[index];
+        if slot.state != SlotState::InUse {
+            return Err(PoolError::NotInUse);
+        }
+
+        slot.uses += 1;
+        if ok {
+            slot.failures = 0;
+            slot.state = SlotState::Idle;
+        } else {
+            slot.failures += 1;
+            if threshold > 0 && slot.failures >= threshold {
+                slot.state = SlotState::Unhealthy;
+                self.stats.marked_unhealthy += 1;
+            } else {
+                slot.state = SlotState::Idle;
+            }
+        }
+        self.stats.released += 1;
+        Ok(())
+    }
+
+    /// Report slot health and repair unhealthy slots by replacing them.
+    pub fn health_check(&mut self, now: u32) -> HealthReport {
+        let mut report = HealthReport::default();
+        for i in 0..self.len {
+            if self.slots[i].state == SlotState::Unhealthy {
+                let standby = self.slots[i].standby;
+                self.slots[i] = Slot::fresh(now, standby);
+                self.stats.recycled += 1;
+            }
+        }
+        for i in 0..self.len {
+            match self.slots[i].state {
+                SlotState::Idle => report.idle += 1,
+                SlotState::InUse => report.in_use += 1,
+                SlotState::Unhealthy => report.unhealthy += 1,
+            }
+            if !self.slots[i].standby && self.slots[i].state != SlotState::Unhealthy {
+                report.healthy = true;
+            }
+        }
+        report
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn config() -> PoolConfig {
+        PoolConfig { capacity: 2, standby: 1, max_age: 100, max_uses: 3, failure_threshold: 2 }
+    }
+
+    #[test]
+    fn rejects_invalid_config() {
+        let mut c = config();
+        c.capacity = 0;
+        assert_eq!(Pool::new(c, 0).err(), Some(PoolError::InvalidConfig));
+        let mut c = config();
+        c.capacity = MAX_POOL_SLOTS;
+        c.standby = 1;
+        assert_eq!(Pool::new(c, 0).err(), Some(PoolError::InvalidConfig));
+    }
+
+    #[test]
+    fn acquires_primary_then_fails_over_then_exhausts() {
+        let mut p = Pool::new(config(), 0).unwrap();
+        assert_eq!(p.acquire(0), Ok(0));
+        assert_eq!(p.acquire(0), Ok(1));
+        assert_eq!(p.stats().failovers, 0);
+        assert_eq!(p.acquire(0), Ok(2));
+        assert_eq!(p.stats().failovers, 1);
+        assert_eq!(p.acquire(0), Err(PoolError::Exhausted));
+        assert_eq!(p.stats().exhausted, 1);
+    }
+
+    #[test]
+    fn released_slot_is_reused() {
+        let mut p = Pool::new(config(), 0).unwrap();
+        let a = p.acquire(0).unwrap();
+        p.release(a, true).unwrap();
+        assert_eq!(p.acquire(0), Ok(a));
+        assert_eq!(p.release(9, true), Err(PoolError::InvalidSlot));
+        let b = p.acquire(0).unwrap();
+        p.release(b, true).unwrap();
+        assert_eq!(p.release(b, true), Err(PoolError::NotInUse));
+    }
+
+    #[test]
+    fn recycles_aged_and_worn_slots() {
+        let mut p = Pool::new(config(), 0).unwrap();
+        // Wear slot 0 out: three uses reaches max_uses.
+        for _ in 0..3 {
+            let s = p.acquire(0).unwrap();
+            assert_eq!(s, 0);
+            p.release(s, true).unwrap();
+        }
+        assert_eq!(p.slot(0).unwrap().uses, 3);
+        assert_eq!(p.recycle(1), 1);
+        assert_eq!(p.slot(0).unwrap().uses, 0);
+
+        // Age every idle slot out.
+        assert_eq!(p.recycle(1_000), 3);
+        assert_eq!(p.slot(1).unwrap().created_at, 1_000);
+        assert_eq!(p.stats().recycled, 4);
+    }
+
+    #[test]
+    fn failing_slot_is_marked_unhealthy_and_failed_over() {
+        let mut p = Pool::new(config(), 0).unwrap();
+        for _ in 0..2 {
+            let s = p.acquire(0).unwrap();
+            assert_eq!(s, 0);
+            p.release(s, false).unwrap();
+        }
+        assert_eq!(p.slot(0).unwrap().state, SlotState::Unhealthy);
+        assert_eq!(p.stats().marked_unhealthy, 1);
+
+        // Slot 1 is next healthy primary, then standby takes over.
+        assert_eq!(p.acquire(0), Ok(1));
+        assert_eq!(p.acquire(0), Ok(2));
+        assert_eq!(p.acquire(0), Err(PoolError::Exhausted));
+    }
+
+    #[test]
+    fn health_check_repairs_unhealthy_slots() {
+        let mut p = Pool::new(config(), 0).unwrap();
+        for _ in 0..2 {
+            let s = p.acquire(0).unwrap();
+            p.release(s, false).unwrap();
+        }
+        let held = p.acquire(0).unwrap();
+        assert_eq!(held, 1);
+
+        let report = p.health_check(10);
+        assert_eq!(report, HealthReport { idle: 2, in_use: 1, unhealthy: 0, healthy: true });
+        assert_eq!(p.slot(0).unwrap().state, SlotState::Idle);
+        assert_eq!(p.slot(0).unwrap().created_at, 10);
+    }
+
+    #[test]
+    fn success_resets_failure_streak() {
+        let mut p = Pool::new(config(), 0).unwrap();
+        let s = p.acquire(0).unwrap();
+        p.release(s, false).unwrap();
+        let s = p.acquire(0).unwrap();
+        p.release(s, true).unwrap();
+        let s = p.acquire(0).unwrap();
+        p.release(s, false).unwrap();
+        assert_eq!(p.slot(0).unwrap().state, SlotState::Idle);
+    }
+}
diff --git a/docs/ISSUE_1023_ERROR_SCENARIOS.md b/docs/ISSUE_1023_ERROR_SCENARIOS.md
new file mode 100644
index 0000000..ddc36a6
--- /dev/null
+++ b/docs/ISSUE_1023_ERROR_SCENARIOS.md
@@ -0,0 +1,33 @@
+# Error scenario testing (#1023)
+
+The suite lives in `contracts/split/src/error_scenario_tests.rs`.
+
+## How a scenario runs
+
+1. A fresh environment is created with an initialized contract and a pending 100-stroop invoice.
+2. **Error injection:** one invalid operation is attempted through the `try_*` client, which must be rejected.
+3. **State check:** the invoice's `funded` and `status` must equal their values from before the attempt.
+4. **Recovery:** a valid payment of the full amount must still succeed and release the invoice.
+
+## Scenario catalogue
+
+| Scenario | Category | Expected result |
+|---|---|---|
+| pay zero | InvalidAmount | rejected |
+| pay negative | InvalidAmount | rejected |
+| pay more than remaining | InvalidAmount | rejected |
+| pay after deadline | Timing | rejected |
+| refund before deadline | Timing | rejected |
+| create invoice with zero amount | InvalidInput | rejected |
+| create invoice with mismatched recipients and amounts | InvalidInput | rejected |
+| pay an unknown invoice | NotFound | rejected |
+
+Two further tests cover state errors: cancelling an invoice that is already released, and refunding twice.
+
+## Reporting and analytics
+
+`ErrorReport` records which scenarios were rejected or wrongly accepted, how many recovered, and a count per `ErrorCategory`. The `error_analytics_group_scenarios_by_category` test checks the per-category totals.
+
+## Adding a scenario
+
+Add an entry to `scenarios()` with a name, a category and an `inject` function that returns `true` when the contract rejects the operation, then update the table above.