monty 0.0.19-beta.2

A sandboxed, snapshotable Python interpreter written in Rust.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
//! Compiled regex pattern type for the `re` module.
//!
//! `RePattern` wraps a compiled `fancy_regex::Regex` with the original Python pattern
//! string and flags. The `fancy_regex` crate supports backreferences, lookahead/lookbehind,
//! and other advanced features, but uses backtracking which means patterns are susceptible
//! to ReDoS. Monty's resource limits (time and allocation budgets) are the primary defense
//! against catastrophic backtracking in untrusted patterns.
//!
//! Custom serde serializes only the pattern string and flags, recompiling the regex
//! on deserialization. This supports Monty's snapshot/restore feature.

use std::{borrow::Cow, cell::OnceCell, cmp::Ordering, fmt::Write, iter, mem, str};

use fancy_regex::{CompileError, Error as RegexError, Regex, RegexBuilder};
use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
use smallvec::SmallVec;

use crate::{
    args::{ArgValues, FromArgs},
    bytecode::{CallResult, VM},
    defer_drop,
    exception_private::{ExcType, RunError, RunResult},
    heap::{Heap, HeapData, HeapId, HeapItem, HeapRead, HeapReadOutput},
    intern::StaticStrings,
    modules::re::{ASCII, DOTALL, IGNORECASE, MULTILINE},
    resource::{ResourceTracker, check_estimated_size},
    types::{
        LazyHeapSet, List, PyTrait, ReMatch, Type, allocate_tuple,
        str::{allocate_string, string_repr_fmt},
    },
    value::{EitherStr, Value},
};

/// A compiled regular expression pattern.
///
/// Wraps a `fancy_regex::Regex` with the original Python pattern string and flags.
/// The `fancy_regex` crate supports backtracking features like backreferences and
/// lookaround, but this means patterns are susceptible to ReDoS — Monty's resource
/// limits are the defense against catastrophic backtracking.
///
/// Custom serde serializes only the pattern string and flags, recompiling the
/// regex on deserialization. This supports Monty's snapshot/restore feature.
#[derive(Debug, Clone)]
pub(crate) struct RePattern {
    /// The original Python regex pattern string.
    pattern: String,
    /// Python regex flags bitmask (IGNORECASE=2, MULTILINE=8, DOTALL=16, ASCII=256).
    flags: u16,
    /// The compiled Rust regex, unanchored.
    compiled: Regex,
    /// The regex anchored with `\A(?:...)` for `match()`, compiled lazily on first
    /// use (most patterns are only ever `search`/`split`/`sub`ed).
    ///
    /// Uses `\A` (absolute start anchor) instead of `^` so the MULTILINE flag
    /// doesn't cause it to match at line boundaries. This correctly handles
    /// alternations — e.g. `match('b|ab', 'ab')` must match `ab`, not fail
    /// because the engine found only `b` starting at position 1.
    compiled_match: OnceCell<Regex>,
    /// The regex anchored with `\A(?:...)\z` for `fullmatch()`, compiled lazily on
    /// first use (see `compiled_match`).
    ///
    /// Uses `\A`/`\z` (absolute anchors) instead of `^`/`$` so the MULTILINE flag
    /// doesn't cause them to match at line boundaries. This correctly handles
    /// alternations — e.g. `fullmatch('a|ab', 'ab')` must match `ab`, not fail
    /// because the engine found `a` first.
    compiled_fullmatch: OnceCell<Regex>,
    /// The `delegate_size_limit` the plain regex was compiled with, forwarded to
    /// the anchored variants above so a *cached* entry's total retained compiled
    /// size stays bounded (the anchors add only O(1) bytes, so a pattern that fit
    /// unanchored still fits anchored). `None` = the engine's default limit. Not
    /// serialized — restored patterns recompile at the default limit.
    delegate_size_limit: Option<usize>,
}

impl PartialEq for RePattern {
    fn eq(&self, other: &Self) -> bool {
        self.pattern == other.pattern && self.flags == other.flags
    }
}

/// Failure of [`RePattern::compile_bounded`], separating "valid pattern whose
/// compiled form exceeds the size cap" (the caller retries uncached at default
/// limits) from a genuine pattern error (reported to the user).
pub(crate) enum BoundedCompileError {
    /// The compiled regex exceeded the requested `delegate_size_limit`.
    TooBig,
    /// The pattern itself is invalid, already converted to `re.PatternError`.
    Invalid(RunError),
}

impl RePattern {
    /// Creates a compiled pattern from a Python regex string and flags.
    ///
    /// Translates Python flag constants into inline regex flag prefixes and compiles
    /// the unanchored pattern. The anchored variants used by `match`/`fullmatch` are
    /// compiled lazily on first use (see [`RePattern::match_regex`]).
    ///
    /// # Errors
    ///
    /// Returns `re.PatternError` if the pattern is invalid.
    pub fn compile(pattern: String, flags: u16) -> RunResult<Self> {
        Self::compile_inner(pattern, flags, None).map_err(ExcType::re_pattern_error)
    }

    /// As [`RePattern::compile`], but caps the compiled size of the delegated regex
    /// (`RegexBuilder::delegate_size_limit`) so the `re` module's pattern cache can
    /// retain entries with a hard per-entry memory ceiling. The limit is retained
    /// and applied to the lazily-compiled anchored `match`/`fullmatch` variants
    /// too, so a cached entry cannot pin large regexes via `.match()`/`.fullmatch()`.
    pub(crate) fn compile_bounded(
        pattern: String,
        flags: u16,
        delegate_size_limit: usize,
    ) -> Result<Self, BoundedCompileError> {
        Self::compile_inner(pattern, flags, Some(delegate_size_limit)).map_err(|err| {
            if is_size_limit_error(&err) {
                BoundedCompileError::TooBig
            } else {
                BoundedCompileError::Invalid(ExcType::re_pattern_error(err))
            }
        })
    }

    /// Shared constructor for [`RePattern::compile`] / [`RePattern::compile_bounded`].
    fn compile_inner(pattern: String, flags: u16, delegate_size_limit: Option<usize>) -> Result<Self, RegexError> {
        let compiled = compile_regex_limited(&pattern, flags, delegate_size_limit)?;
        Ok(Self {
            pattern,
            flags,
            compiled,
            compiled_match: OnceCell::new(),
            compiled_fullmatch: OnceCell::new(),
            delegate_size_limit,
        })
    }

    /// Returns the `\A(?:pattern)` regex for `match()`, compiling it on first use.
    ///
    /// Wrapping a pattern that already compiled essentially never fails, so any
    /// error surfaces (as `re.PatternError`) at `match()` rather than `re.compile()`.
    fn match_regex(&self) -> RunResult<&Regex> {
        if let Some(regex) = self.compiled_match.get() {
            return Ok(regex);
        }
        let compiled = compile_regex_limited(
            &format!("\\A(?:{})", self.pattern),
            self.flags,
            self.delegate_size_limit,
        )
        .map_err(ExcType::re_pattern_error)?;
        // `set` only fails on a concurrent init, impossible on the single-threaded VM.
        let _ = self.compiled_match.set(compiled);
        Ok(self.compiled_match.get().expect("cell was just initialised"))
    }

    /// Returns the `\A(?:pattern)\z` regex for `fullmatch()`, compiling on first use.
    fn fullmatch_regex(&self) -> RunResult<&Regex> {
        if let Some(regex) = self.compiled_fullmatch.get() {
            return Ok(regex);
        }
        let compiled = compile_regex_limited(
            &format!("\\A(?:{})\\z", self.pattern),
            self.flags,
            self.delegate_size_limit,
        )
        .map_err(ExcType::re_pattern_error)?;
        let _ = self.compiled_fullmatch.set(compiled);
        Ok(self.compiled_fullmatch.get().expect("cell was just initialised"))
    }

    /// Builds a single `ReMatch` heap value from a capture result, keeping the
    /// subject alive by refcount (`subject.clone_with_heap`) rather than copying
    /// its text. `all_ascii` is precomputed by the caller (once per `finditer`).
    fn build_match(
        &self,
        caps: &fancy_regex::Captures<'_>,
        subject: &Value,
        all_ascii: bool,
        heap: &Heap<impl ResourceTracker>,
    ) -> RunResult<Value> {
        let m = ReMatch::from_captures(caps, subject.clone_with_heap(heap), all_ascii, &self.compiled);
        Ok(Value::Ref(heap.allocate(HeapData::ReMatch(m))?))
    }

    /// `pattern.search(string)` — find first match anywhere in the string.
    ///
    /// `subject` is the subject `Value` (stored by the match); `text` is its
    /// borrowed contents. Returns a `ReMatch` heap object, or `Value::None`.
    pub fn search(&self, subject: &Value, text: &str, heap: &Heap<impl ResourceTracker>) -> RunResult<Value> {
        match self.compiled.captures(text) {
            Ok(Some(caps)) => self.build_match(&caps, subject, text.is_ascii(), heap),
            Ok(None) => Ok(Value::None),
            Err(err) => Err(ExcType::re_pattern_error(err)),
        }
    }

    /// `pattern.match(string)` — match anchored at the start of the string.
    ///
    /// Uses a pre-compiled `\A(?:pattern)` regex to correctly handle alternations.
    /// For example, `match('b|ab', 'ab')` correctly matches `ab` because the
    /// anchor forces the engine to try all alternatives at position 0.
    ///
    /// Returns a `ReMatch` heap object on success, or `Value::None` if no match.
    pub fn match_start(&self, subject: &Value, text: &str, heap: &Heap<impl ResourceTracker>) -> RunResult<Value> {
        match self.match_regex()?.captures(text) {
            Ok(Some(caps)) => self.build_match(&caps, subject, text.is_ascii(), heap),
            Ok(None) => Ok(Value::None),
            Err(err) => Err(ExcType::re_pattern_error(err)),
        }
    }

    /// `pattern.fullmatch(string)` — match the entire string.
    ///
    /// Uses a pre-compiled `\A(?:pattern)\z` regex to correctly handle alternations.
    /// For example, `fullmatch('a|ab', 'ab')` correctly matches `ab` because the
    /// anchors force the engine to try all alternatives for a full-string match.
    ///
    /// Returns a `ReMatch` heap object on success, or `Value::None` if no match.
    pub fn fullmatch(&self, subject: &Value, text: &str, heap: &Heap<impl ResourceTracker>) -> RunResult<Value> {
        match self.fullmatch_regex()?.captures(text) {
            Ok(Some(caps)) => self.build_match(&caps, subject, text.is_ascii(), heap),
            Ok(None) => Ok(Value::None),
            Err(err) => Err(ExcType::re_pattern_error(err)),
        }
    }

    /// `pattern.findall(string)` — return all non-overlapping matches.
    ///
    /// Follows CPython's semantics:
    /// - No capture groups: returns a list of matched strings
    /// - One capture group: returns a list of the group's matched strings
    /// - Multiple capture groups: returns a list of tuples of matched strings
    pub fn findall(&self, text: &str, heap: &Heap<impl ResourceTracker>) -> RunResult<Value> {
        let cap_count = self.compiled.captures_len();
        let mut results = Vec::new();

        match cap_count {
            // No capture groups — return list of full match strings
            0 | 1 => {
                for m in self.compiled.find_iter(text) {
                    let val = m.map_err(ExcType::re_pattern_error)?.as_str();
                    results.push(allocate_string(val, heap)?);
                }
            }
            // One capture group — return list of the group's strings
            2 => {
                for caps in self.compiled.captures_iter(text) {
                    let caps = caps.map_err(ExcType::re_pattern_error)?;
                    let val = caps.get(1).map_or("", |m| m.as_str());
                    results.push(allocate_string(val, heap)?);
                }
            }
            // Multiple capture groups — return list of tuples
            _ => {
                for caps in self.compiled.captures_iter(text) {
                    let caps = caps.map_err(ExcType::re_pattern_error)?;
                    let mut elements: SmallVec<[Value; 3]> = SmallVec::with_capacity(cap_count - 1);
                    for cap in caps.iter().skip(1) {
                        let val = cap.map_or("", |m| m.as_str());
                        elements.push(allocate_string(val, heap)?);
                    }
                    results.push(allocate_tuple(elements, heap)?);
                }
            }
        }

        let list = List::new(results);
        Ok(Value::Ref(heap.allocate(HeapData::List(list))?))
    }

    /// `pattern.sub(repl, string, count=0)` — substitute matches with a replacement.
    ///
    /// When `count` is 0, all matches are replaced. Otherwise, at most `count`
    /// replacements are made. The replacement string supports `$1`, `$2`, etc.
    /// for backreferences to captured groups.
    ///
    /// Builds the result string in a single pass by iterating matches and appending
    /// replacements directly. Checks the running output size against resource limits
    /// after each match, bailing out immediately if the budget is exceeded. This
    /// avoids both false rejections from conservative pre-estimates and untracked
    /// Rust heap allocations from delegating to `fancy_regex::replace_all()`.
    pub fn sub(&self, repl: &str, text: &str, count: usize, heap: &Heap<impl ResourceTracker>) -> RunResult<Value> {
        // Translate Python-style backreferences (\1, \2) to regex crate style ($1, $2)
        let rust_repl = translate_replacement(repl);
        let effective_count = if count == 0 { usize::MAX } else { count };

        let mut result = String::new();
        let mut last_end = 0;

        for caps in self.compiled.captures_iter(text).take(effective_count) {
            let caps = caps.map_err(ExcType::re_pattern_error)?;
            let m = caps.get(0).expect("capture group 0 always exists");
            result.push_str(&text[last_end..m.start()]);
            caps.expand(rust_repl.as_ref(), &mut result);
            last_end = m.end();
            // Check running size: current result + remaining unprocessed text.
            check_estimated_size(result.len() + (text.len() - last_end), heap.tracker())?;
        }

        result.push_str(&text[last_end..]);
        Ok(allocate_string(result, heap)?)
    }

    /// `pattern.split(string, maxsplit=0)` — split string by pattern occurrences.
    ///
    /// Returns a list of strings. If `maxsplit` is positive, at most `maxsplit`
    /// splits occur and the remainder of the string is returned as the final
    /// element; if it is negative, no splits occur at all (CPython's split loop
    /// runs zero times), returning the whole subject as a single element.
    pub fn split(&self, text: &str, maxsplit: i64, heap: &Heap<impl ResourceTracker>) -> RunResult<Value> {
        let pieces: Vec<&str> = match maxsplit.cmp(&0) {
            Ordering::Less => vec![text],
            Ordering::Equal => self
                .compiled
                .split(text)
                .collect::<Result<Vec<_>, _>>()
                .map_err(ExcType::re_pattern_error)?,
            Ordering::Greater => {
                // `maxsplit + 1` pieces = at most `maxsplit` splits; saturate
                // for absurdly large limits (splitn caps at the piece count).
                let limit = usize::try_from(maxsplit).unwrap_or(usize::MAX).saturating_add(1);
                self.compiled
                    .splitn(text, limit)
                    .collect::<Result<Vec<_>, _>>()
                    .map_err(ExcType::re_pattern_error)?
            }
        };

        let mut results = Vec::with_capacity(pieces.len());
        for piece in pieces {
            results.push(allocate_string(piece, heap)?);
        }

        let list = List::new(results);
        Ok(Value::Ref(heap.allocate(HeapData::List(list))?))
    }

    /// `pattern.finditer(string)` — return all matches as a list.
    ///
    /// Eagerly collects all match objects into a list. This differs from CPython's
    /// lazy iterator but produces the same results when iterated. The VM's `GetIter`
    /// opcode handles iteration over the returned list.
    pub fn finditer(&self, subject: &Value, text: &str, heap: &Heap<impl ResourceTracker>) -> RunResult<Value> {
        // Every match shares one refcounted subject reference, not a copy each.
        let all_ascii = text.is_ascii();

        let mut results = Vec::new();
        for caps in self.compiled.captures_iter(text) {
            let caps = caps.map_err(ExcType::re_pattern_error)?;
            results.push(self.build_match(&caps, subject, all_ascii, heap)?);
        }

        let list = List::new(results);
        Ok(Value::Ref(heap.allocate(HeapData::List(list))?))
    }
}

impl<'h> PyTrait<'h> for HeapRead<'h, RePattern> {
    fn py_type(&self, _vm: &VM<'h, impl ResourceTracker>) -> Type {
        Type::RePattern
    }

    fn py_len(&self, _vm: &VM<'h, impl ResourceTracker>) -> Option<usize> {
        None
    }

    fn py_eq_impl(&self, other: &Value, vm: &mut VM<'h, impl ResourceTracker>) -> RunResult<Option<bool>> {
        let Some(HeapReadOutput::RePattern(other)) = other.read_heap(vm) else {
            return Ok(None);
        };
        Ok(Some(self.get(vm.heap) == other.get(vm.heap)))
    }

    fn py_bool(&self, _vm: &mut VM<'h, impl ResourceTracker>) -> bool {
        // Pattern objects are always truthy (matching CPython).
        true
    }

    fn py_repr_fmt(
        &self,
        f: &mut impl Write,
        vm: &mut VM<'h, impl ResourceTracker>,
        _heap_ids: &mut LazyHeapSet,
    ) -> RunResult<()> {
        let this = self.get(vm.heap);
        write!(f, "re.compile(")?;
        string_repr_fmt(&this.pattern, f)?;
        if this.flags != 0 {
            let mut flag_parts = smallvec::SmallVec::<[&'static str; 4]>::new();
            if this.flags & IGNORECASE != 0 {
                flag_parts.push("re.IGNORECASE");
            }
            if this.flags & MULTILINE != 0 {
                flag_parts.push("re.MULTILINE");
            }
            if this.flags & DOTALL != 0 {
                flag_parts.push("re.DOTALL");
            }
            if this.flags & ASCII != 0 {
                flag_parts.push("re.ASCII");
            }
            write!(f, ", {}", flag_parts.join("|"))?;
        }
        Ok(write!(f, ")")?)
    }

    fn py_getattr(&self, attr: &EitherStr, vm: &mut VM<'h, impl ResourceTracker>) -> RunResult<Option<CallResult>> {
        match attr.static_string() {
            Some(StaticStrings::PatternAttr) => {
                let v = allocate_string(self.get(vm.heap).pattern.as_str(), vm.heap)?;
                Ok(Some(CallResult::Value(v)))
            }
            Some(StaticStrings::Flags) => Ok(Some(CallResult::Value(Value::Int(i64::from(self.get(vm.heap).flags))))),
            _ => Err(ExcType::attribute_error(Type::RePattern, attr.as_str(vm.interns))),
        }
    }

    fn py_call_attr(
        &mut self,
        _self_id: HeapId,
        vm: &mut VM<'h, impl ResourceTracker>,
        attr: &EitherStr,
        args: ArgValues,
    ) -> RunResult<CallResult> {
        let result = match attr.static_string() {
            Some(StaticStrings::Search) => {
                let arg = args.get_one_arg("Pattern.search", vm.heap)?;
                defer_drop!(arg, vm);
                let text = arg.to_str(vm)?;
                self.get(vm.heap).search(arg, text, vm.heap)
            }
            Some(StaticStrings::Match) => {
                let arg = args.get_one_arg("Pattern.match", vm.heap)?;
                defer_drop!(arg, vm);
                let text = arg.to_str(vm)?;
                self.get(vm.heap).match_start(arg, text, vm.heap)
            }
            Some(StaticStrings::Fullmatch) => {
                let arg = args.get_one_arg("Pattern.fullmatch", vm.heap)?;
                defer_drop!(arg, vm);
                let text = arg.to_str(vm)?;
                self.get(vm.heap).fullmatch(arg, text, vm.heap)
            }
            Some(StaticStrings::Findall) => {
                let arg = args.get_one_arg("Pattern.findall", vm.heap)?;
                defer_drop!(arg, vm);
                let text = arg.to_str(vm)?;
                self.get(vm.heap).findall(text, vm.heap)
            }
            Some(StaticStrings::Sub) => call_pattern_sub(self, args, vm),
            Some(StaticStrings::Split) => call_pattern_split(self, args, vm),
            Some(StaticStrings::Finditer) => {
                let arg = args.get_one_arg("Pattern.finditer", vm.heap)?;
                defer_drop!(arg, vm);
                let text = arg.to_str(vm)?;
                self.get(vm.heap).finditer(arg, text, vm.heap)
            }
            _ => {
                return Err(ExcType::attribute_error(Type::RePattern, attr.as_str(vm.interns)));
            }
        }?;
        Ok(CallResult::Value(result))
    }
}

impl HeapItem for RePattern {
    fn py_estimate_size(&self) -> usize {
        mem::size_of::<Self>() + self.pattern.len()
    }

    fn py_dec_ref_ids(&mut self, _stack: &mut Vec<HeapId>) {
        // No heap references — all data is owned.
    }
}

/// Handles `pattern.sub(repl, string, count=0)` argument extraction and dispatch.
///
/// Separated from the main `py_call_attr` match to keep the borrow checker happy —
/// extracting multiple string arguments requires careful ordering of borrows.
/// Supports `count` as either positional or keyword argument.
fn call_pattern_sub<'h>(
    pattern: &HeapRead<'h, RePattern>,
    args: ArgValues,
    vm: &mut VM<'h, impl ResourceTracker>,
) -> RunResult<Value> {
    let PatternSubArgs {
        repl: repl_val,
        string: string_val,
        count: count_val,
    } = PatternSubArgs::from_args(args, vm)?;
    defer_drop!(repl_val, vm);
    defer_drop!(string_val, vm);

    let count = extract_count(count_val, vm)?;

    // Check that repl is a string — callable replacement is not supported.
    // CPython processes the replacement template *before* its match loop, so
    // this check must precede the negative-count early return below: a bad
    // repl raises even when zero substitutions will run.
    if !repl_val.is_str(vm.heap) {
        return Err(ExcType::type_error(
            "callable replacement is not yet supported in re.sub()",
        ));
    }

    let Some(count) = count else {
        // Negative count — Pattern.sub returns the input string unchanged.
        // The subject is still type-checked (`to_str` raises this method's
        // `expected string, not {t}` wording) before the refcount bump; no
        // need to re-allocate.
        let _ = string_val.to_str(vm)?;
        return Ok(string_val.clone_with_heap(vm.heap));
    };

    let repl = repl_val.to_str(vm)?.to_owned();
    let text = string_val.to_str(vm)?.to_owned();
    pattern.get(vm.heap).sub(&repl, &text, count, vm.heap)
}

/// Handles `pattern.split(string, maxsplit=0)` argument extraction and dispatch.
///
/// Supports `maxsplit` as either positional or keyword argument.
fn call_pattern_split<'h>(
    pattern: &HeapRead<'h, RePattern>,
    args: ArgValues,
    vm: &mut VM<'h, impl ResourceTracker>,
) -> RunResult<Value> {
    let PatternSplitArgs {
        string: string_val,
        maxsplit: maxsplit_val,
    } = PatternSplitArgs::from_args(args, vm)?;
    defer_drop!(string_val, vm);

    let maxsplit = extract_maxsplit(maxsplit_val, vm)?;
    let text = string_val.to_str(vm)?.to_owned();
    pattern.get(vm.heap).split(&text, maxsplit, vm.heap)
}

/// Argument shape for `Pattern.sub(repl, string, count=0)`.
///
/// `string` uses `static_string = "StringAttr"` because `StringAttr` is the
/// `StaticStrings` entry that interns `"string"` (the bare `String` variant
/// is taken by the `re.Pattern.string` attribute name in CPython's class
/// hierarchy).
#[derive(FromArgs)]
#[from_args(name = "sub", style = c_named, at_most_total)]
struct PatternSubArgs {
    repl: Value,
    #[from_args(static_string = "StringAttr")]
    string: Value,
    #[from_args(default)]
    count: Option<Value>,
}

/// Argument shape for `Pattern.split(string, maxsplit=0)`.
///
/// See `PatternSubArgs` for why `string` uses `static_string`.
#[derive(FromArgs)]
#[from_args(name = "split", style = c_named, at_most_total)]
struct PatternSplitArgs {
    #[from_args(static_string = "StringAttr")]
    string: Value,
    #[from_args(default)]
    maxsplit: Option<Value>,
}

/// Extracts a `maxsplit` value from an optional `Value` for [`RePattern::split`].
///
/// Returns 0 (split all) if not provided; negatives pass through — the split
/// loop then runs zero times, matching CPython. Non-ints get CPython's
/// argument-clinic message. Shared by `Pattern.split` and module-level
/// `re.split`.
pub(crate) fn extract_maxsplit(val: Option<Value>, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<i64> {
    match val {
        None => Ok(0),
        Some(Value::Int(n)) => Ok(n),
        Some(Value::Bool(b)) => Ok(i64::from(b)),
        Some(other) => {
            let t = other.py_type_name(vm);
            other.drop_with_heap(vm);
            Err(ExcType::type_error(format!(
                "'{t}' object cannot be interpreted as an integer"
            )))
        }
    }
}

/// Extracts a `count` value from an optional `Value` for [`RePattern::sub`].
///
/// Returns `Ok(None)` for a negative count, which callers turn into "return
/// the subject unchanged" (CPython's match loop runs zero times there).
/// Non-ints get CPython's argument-clinic message. Shared by `Pattern.sub`
/// and module-level `re.sub`.
pub(crate) fn extract_count(val: Option<Value>, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<Option<usize>> {
    match val {
        None => Ok(Some(0)),
        // Saturate rather than `as`-cast: on 32-bit targets (wasm) a count
        // above usize::MAX would otherwise truncate — e.g. 2**32 to 0, which
        // means "replace all" instead of an unreachably large cap.
        Some(Value::Int(n)) if n >= 0 => Ok(Some(usize::try_from(n).unwrap_or(usize::MAX))),
        Some(Value::Bool(b)) => Ok(Some(usize::from(b))),
        Some(Value::Int(_)) => Ok(None),
        Some(other) => {
            let t = other.py_type_name(vm);
            other.drop_with_heap(vm);
            Err(ExcType::type_error(format!(
                "'{t}' object cannot be interpreted as an integer"
            )))
        }
    }
}

/// Compiles a Python regex pattern string with flags into a Rust `Regex`.
///
/// Translates Python flag constants into inline regex flag prefixes:
/// - `re.IGNORECASE` (2) → `(?i)` prefix
/// - `re.MULTILINE` (8) → `(?m)` prefix
/// - `re.DOTALL` (16) → `(?s)` prefix
///
/// `delegate_size_limit` optionally caps the compiled size of the delegated
/// regex (`RegexBuilder::delegate_size_limit`) — used to bound cached patterns;
/// `None` uses the engine's default limit.
///
/// # Errors
///
/// Returns the raw `fancy_regex` error so callers can distinguish a size-limit
/// overflow from an invalid pattern (see [`is_size_limit_error`]) before
/// converting to `re.PatternError`.
fn compile_regex_limited(pattern: &str, flags: u16, delegate_size_limit: Option<usize>) -> Result<Regex, RegexError> {
    let mut prefix = String::new();
    if flags & IGNORECASE != 0 {
        prefix.push('i');
    }
    if flags & MULTILINE != 0 {
        prefix.push('m');
    }
    if flags & DOTALL != 0 {
        prefix.push('s');
    }
    // Note: re.ASCII (256) is accepted but has no effect on the regex compilation.
    // `fancy_regex` doesn't support `(?-u)` to disable Unicode mode, so `\w`, `\d`, `\s`
    // always match Unicode characters. This is a known limitation — Python 3 defaults to
    // Unicode mode anyway, so the behavioral difference only matters for non-ASCII input.

    let full_pattern = if prefix.is_empty() {
        pattern.to_owned()
    } else {
        format!("(?{prefix}){pattern}")
    };

    let mut builder = RegexBuilder::new(&full_pattern);
    if let Some(limit) = delegate_size_limit {
        builder.delegate_size_limit(limit);
    }
    builder.build()
}

/// True when `err` is the delegated engine's exceeded-size-limit error: the
/// pattern is valid, its compiled form just doesn't fit the requested cap.
fn is_size_limit_error(err: &RegexError) -> bool {
    match err {
        RegexError::CompileError(compile_error) => match &**compile_error {
            CompileError::InnerError(inner) => inner.size_limit().is_some(),
            _ => false,
        },
        _ => false,
    }
}

/// Translates Python-style replacement backreferences to `fancy_regex` syntax.
///
/// Python uses `\1`, `\2`, `\g<1>`, `\g<name>` for backreferences in replacement strings.
/// `fancy_regex` uses `$1`, `$2`, `${1}`, `${name}`. This function converts between them.
///
/// # Supported translations
///
/// - `\1`–`\9` → `$1`–`$9` (single-digit backreferences)
/// - `\g<N>` → `${N}` (numeric backreference with explicit syntax)
/// - `\g<name>` → `${name}` (named group backreference)
/// - `\\` → literal backslash
/// - `$` → `$$` (escape literal `$` so `fancy_regex` doesn't misinterpret it)
///
/// Returns a `Cow` to avoid allocation when no translation is needed.
///
/// # Limitations
///
/// TODO: Multi-digit backreferences like `\10` are not fully supported. CPython
/// greedily reads all digits after `\` and interprets them as a group number if
/// that group exists, otherwise falls back to octal escapes. Currently `\10` is
/// translated as `$1` followed by literal `0`, which is wrong when 10+ groups
/// exist. Fixing this requires passing the pattern's capture group count into
/// this function to disambiguate.
fn translate_replacement(repl: &str) -> Cow<'_, str> {
    // Fast path: no backslashes and no literal `$` means nothing to translate or escape.
    if !repl.contains('\\') && !repl.contains('$') {
        return Cow::Borrowed(repl);
    }

    let mut result = String::with_capacity(repl.len());
    let mut chars = repl.chars().peekable();

    while let Some(c) = chars.next() {
        if c == '\\' {
            match chars.peek() {
                Some(&d) if d.is_ascii_digit() => {
                    // TODO: This only handles single-digit backrefs (\1–\9).
                    // Multi-digit like \10 should be ${10} when group 10 exists,
                    // but that requires knowing the group count. See docstring.
                    result.push('$');
                    result.push(d);
                    chars.next();
                }
                Some(&'g') => {
                    chars.next(); // consume 'g'
                    translate_g_backref(&mut chars, &mut result);
                }
                Some(&'\\') => {
                    result.push('\\');
                    chars.next();
                }
                _ => {
                    result.push('\\');
                }
            }
        } else if c == '$' {
            // Escape literal `$` as `$$` so `fancy_regex` doesn't interpret `$1` etc.
            // as backreferences.
            result.push('$');
            result.push('$');
        } else {
            result.push(c);
        }
    }

    Cow::Owned(result)
}

/// Translates a `\g<...>` backreference to `fancy_regex` `${...}` syntax.
///
/// Called after `\g` has been consumed. Reads `<name_or_number>` from the iterator
/// and writes `${name_or_number}` to the result. If the syntax is malformed
/// (missing `<` or `>`), the literal characters are written through unchanged.
fn translate_g_backref(chars: &mut iter::Peekable<str::Chars<'_>>, result: &mut String) {
    if chars.peek() != Some(&'<') {
        // Not \g<...>, just literal \g
        result.push('\\');
        result.push('g');
        return;
    }
    chars.next(); // consume '<'

    // Collect everything until '>'
    let mut name = String::new();
    loop {
        match chars.next() {
            Some('>') => break,
            Some(ch) => name.push(ch),
            None => {
                // Unterminated \g<... — emit literally
                result.push('\\');
                result.push('g');
                result.push('<');
                result.push_str(&name);
                return;
            }
        }
    }

    // Write as ${name_or_number} for fancy_regex
    result.push('$');
    result.push('{');
    result.push_str(&name);
    result.push('}');
}

impl Serialize for RePattern {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        // Serialize only pattern string and flags; regex is recompiled on deserialize.
        (&self.pattern, self.flags).serialize(serializer)
    }
}

impl<'de> Deserialize<'de> for RePattern {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let (pattern, flags): (String, u16) = Deserialize::deserialize(deserializer)?;
        Self::compile(pattern, flags).map_err(|e| de::Error::custom(format!("{e:?}")))
    }
}