feat(sms): pick a CSV, pace the batch, and let the user stop it

Wires the previous two commits into the bulk compose tab. Three
user-visible additions.

1. "Import recipients from CSV" opens the system file picker.

   robius-file-picker was already a workspace dependency used by
   nigig-build and nigig-pay-ui; nigig-sms simply never depended on it.
   Same pin. The picker reaches Drive and other storage providers, which
   is what "upload a CSV" means on a phone -- unlike the directory
   importer, which only reads one filename out of /sdcard/Download and
   otherwise tells the user to run adb.

   The callback runs off the UI thread, so it cannot touch Cx. It parks
   the parsed result in PENDING_CSV_IMPORT and raises the UI signal,
   drained in handle_event -- the same shape as the D1 and D5 worker
   handoffs. A cancelled picker leaves the existing list untouched.

   The status line names the first three skipped lines and their
   reasons. "3 rows skipped" is not actionable on a 900-row file.

2. A "seconds between messages" field, defaulting to 60.

   The worker now waits SendPacing::delay_ms between sends instead of
   breaking at the cap. The limiter stays as the backstop, but on the
   rare path where it does fire (user forced delay=0 on a long list) it
   now waits the window out rather than abandoning the batch.

   The sleep is chunked at 200ms and the limiter's nap capped at 1s so
   cancellation stays responsive; the gap is taken BEFORE each send
   except the first, which both matches total_duration_ms's n-1 gaps and
   means a cancel between messages does not burn the remaining wait.

   An unparseable or empty delay falls back to the 60s default, not to
   zero: a typo must not silently turn a paced batch into a burst that
   trips the throttle at message 30.

3. The send button becomes Stop while a batch is running.

   Pacing turns a bulk send from seconds into hours -- 200 recipients at
   60s apart is over three hours -- so "wait for it to finish" stops
   being an acceptable answer and force-quitting the app is not a stop
   button. BULK_SEND_CANCEL is checked while sleeping, not only between
   sends.

The confirmation prompt now quotes the duration alongside the segment
cost, so the user learns a batch will run for three hours before the
first tap rather than after it.

Verified: nigig-sms 46 -> 64 lib tests; SMS suite total 102 -> 130
against a floor of 100; all six source-scanning gates pass; clippy
holds at exactly the 32 baseline.

The slice gate caught a real regression here -- detect_columns used
rows[..sample], which the gate flags on shape. Vec slices are safe, but
rewritten as .take(SAMPLE) rather than raising the baseline.

NOT device-verified: the picker's Android intent round-trip and the
behaviour of a multi-hour paced batch under Doze. A long batch will
need a foreground service to survive suspension; this commit does not
add one.

Cargo.lock

@@ -2724,6 +2724,7 @@ dependencies = [
  "nigig-uikit",
  "proptest",
  "robius-contacts",
+ "robius-file-picker",
  "robius-sms",
 ]
 

crates/apps/nigig-sms/Cargo.toml

@@ -10,6 +10,9 @@ nigig-uikit = { path = "../../nigig-uikit" }
 chrono = { version = "0.4", features = ["serde"] }
 robius-sms = { version = "0.1.0", path = "../../robius-sms" }
 robius-contacts = { version = "0.1.0", path = "../../robius-contacts" }
+# Same pin as nigig-build and nigig-pay-ui, which already use it to
+# pick a CSV. Bulk SMS needs the identical capability.
+robius-file-picker = { git = "https://github.com/project-robius/robius", rev = "b766e62b0600f5d2ee21cc6995648346fc277bd8" }
 
 [dev-dependencies]
 proptest = "1"

crates/apps/nigig-sms/src/sms_frame/bulk/mod.rs

@@ -1,6 +1,7 @@
 pub mod sms_bulk_page;
 pub mod directory_scraper;
 pub mod companies_list;
+pub mod recipient_csv;
 
 use makepad_widgets::ScriptVm;
 

crates/apps/nigig-sms/src/sms_frame/bulk/recipient_csv.rs

@@ -0,0 +1,460 @@
+// Parse an arbitrary user-supplied CSV into a recipient list.
+//
+// This is deliberately NOT `import_business_listings_from_csv_path`.
+// That importer exists for one specific artefact: an 11-column export
+// (`category,company_name,address,phones,emails,industry,source_url,
+// page_number,website,is_favorite,notes`) with a hardcoded filename,
+// discovered by probing /sdcard/Download and three dev paths. It fails
+// the whole file on the first malformed row, and its own error message
+// tells the user to run `adb push`. None of that is reachable for
+// someone who just wants to text a list of numbers they exported from
+// a spreadsheet.
+//
+// What people actually have is a phone column, maybe a name column,
+// maybe a header, and inevitably some junk rows. So:
+//
+//   * find the phone column by NAME if there is a header, otherwise by
+//     content -- the first column where most values look like numbers;
+//   * skip bad rows and report how many, rather than rejecting the file;
+//   * de-duplicate, because sending the same person four copies of a
+//     marketing message is worse than not sending at all;
+//   * keep every row's original line number so the UI can point at what
+//     it skipped.
+//
+// Kept free of Makepad and of any file I/O so it is testable on the
+// host: the SMS crates' CI runs on Linux, where the whole Android
+// backend is a stub, and an untested parser is exactly how bug A3
+// (byte-offset slicing) shipped.
+
+/// One accepted row.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct CsvRecipient {
+    /// Normalised E.164-ish number, ready to hand to the SMS provider.
+    pub phone: String,
+    /// Display name if the file had one, else empty.
+    pub name: String,
+}
+
+/// Outcome of parsing a file: what we can send to, and what we ignored.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct CsvImport {
+    pub recipients: Vec<CsvRecipient>,
+    /// `(line_number, reason)` for each row we could not use.
+    pub skipped: Vec<(usize, String)>,
+    /// Rows dropped because the number already appeared earlier.
+    pub duplicates: usize,
+}
+
+impl CsvImport {
+    pub fn is_empty(&self) -> bool {
+        self.recipients.is_empty()
+    }
+
+    /// One-line summary for the status label.
+    pub fn summary(&self) -> String {
+        let mut s = format!(
+            "{} recipient{}",
+            self.recipients.len(),
+            if self.recipients.len() == 1 { "" } else { "s" }
+        );
+        if self.duplicates > 0 {
+            s.push_str(&format!(", {} duplicate(s) removed", self.duplicates));
+        }
+        if !self.skipped.is_empty() {
+            s.push_str(&format!(", {} row(s) skipped", self.skipped.len()));
+        }
+        s
+    }
+}
+
+/// Header names we accept for the phone column, lowercased.
+const PHONE_HEADERS: &[&str] = &[
+    "phone", "phones", "number", "msisdn", "mobile", "cell", "telephone",
+    "tel", "contact", "phone_number", "phonenumber", "mobile_number",
+];
+
+/// Header names we accept for the display-name column, lowercased.
+const NAME_HEADERS: &[&str] = &[
+    "name", "company_name", "company", "customer", "contact_name",
+    "full_name", "fullname", "business", "client",
+];
+
+/// Split one CSV line, honouring double-quoted fields.
+///
+/// Not a general CSV implementation -- no embedded newlines -- but it
+/// does handle `"Acme, Inc.",+254...`, which a naive `split(',')` gets
+/// wrong and which is extremely common in spreadsheet exports.
+fn split_row(line: &str) -> Vec<String> {
+    let mut out = Vec::new();
+    let mut cur = String::new();
+    let mut in_quotes = false;
+    let mut chars = line.chars().peekable();
+
+    while let Some(c) = chars.next() {
+        match c {
+            '"' if in_quotes && chars.peek() == Some(&'"') => {
+                // Escaped quote inside a quoted field.
+                cur.push('"');
+                chars.next();
+            }
+            '"' => in_quotes = !in_quotes,
+            ',' | ';' | '\t' if !in_quotes => {
+                out.push(cur.trim().to_string());
+                cur = String::new();
+            }
+            _ => cur.push(c),
+        }
+    }
+    out.push(cur.trim().to_string());
+    out
+}
+
+/// Normalise a phone number, or return None if it cannot be one.
+///
+/// Kenyan-friendly (this app's directory is Nairobi businesses) but not
+/// Kenya-only: anything already in `+<country><number>` form is kept.
+///
+///   0712345678      -> +254712345678
+///   254712345678    -> +254712345678
+///   +254 712 345678 -> +254712345678
+///   712345678       -> +254712345678
+///
+/// Returns None for anything too short, too long, or non-numeric, so a
+/// stray "N/A" or an email address in the phone column is skipped
+/// rather than dispatched to the radio.
+pub fn normalise_phone(raw: &str) -> Option<String> {
+    let trimmed = raw.trim();
+    if trimmed.is_empty() {
+        return None;
+    }
+
+    let had_plus = trimmed.starts_with('+');
+    // Strip formatting: spaces, dashes, parens, dots.
+    let digits: String = trimmed.chars().filter(|c| c.is_ascii_digit()).collect();
+    if digits.is_empty() {
+        return None;
+    }
+
+    // A number with letters mixed in is not a number -- reject rather
+    // than silently salvaging the digits out of "call 0712345678 ext 4".
+    let stripped: String = trimmed
+        .chars()
+        .filter(|c| !c.is_whitespace() && !"+-().".contains(*c))
+        .collect();
+    if !stripped.chars().all(|c| c.is_ascii_digit()) {
+        return None;
+    }
+
+    let normalised = if had_plus {
+        // Already international; trust it.
+        format!("+{digits}")
+    } else if let Some(rest) = digits.strip_prefix("254") {
+        format!("+254{rest}")
+    } else if let Some(rest) = digits.strip_prefix('0') {
+        format!("+254{rest}")
+    } else if digits.len() == 9 {
+        // Bare subscriber number, e.g. 712345678.
+        format!("+254{digits}")
+    } else {
+        // Unknown shape with no country code. Keep the digits but do
+        // not invent a country -- an 11-digit US number must not become
+        // a Kenyan one.
+        format!("+{digits}")
+    };
+
+    // Sanity bounds. E.164 allows at most 15 digits; anything under 7
+    // is not dialable.
+    let n = normalised.chars().filter(|c| c.is_ascii_digit()).count();
+    if !(7..=15).contains(&n) {
+        return None;
+    }
+    Some(normalised)
+}
+
+/// Does this cell look like a phone number?
+fn looks_like_phone(cell: &str) -> bool {
+    normalise_phone(cell).is_some()
+}
+
+/// Decide which column holds phones and which holds names.
+///
+/// Returns `(phone_idx, name_idx, header_consumed)`.
+fn detect_columns(rows: &[Vec<String>]) -> (usize, Option<usize>, bool) {
+    let Some(first) = rows.first() else {
+        return (0, None, false);
+    };
+
+    // A header row is one where the phone column is named and the cell
+    // is NOT itself a phone number. "phone" is a header; "+254..." is
+    // data even if a column is called that.
+    let lowered: Vec<String> = first.iter().map(|c| c.to_lowercase()).collect();
+    let header_phone = lowered
+        .iter()
+        .position(|c| PHONE_HEADERS.contains(&c.as_str()));
+
+    if let Some(pi) = header_phone {
+        if !looks_like_phone(&first[pi]) {
+            let ni = lowered
+                .iter()
+                .position(|c| NAME_HEADERS.contains(&c.as_str()));
+            return (pi, ni, true);
+        }
+    }
+
+    // No usable header. Pick the column with the most phone-looking
+    // values across the sample.
+    let width = rows.iter().map(|r| r.len()).max().unwrap_or(1);
+    // `take(50)` rather than `rows[..sample]`: the SMS crates gate
+    // against byte-offset indexing because `s[..n]` on a &str panics on
+    // a non-boundary (bug A3). These are Vec slices and would be safe,
+    // but the gate matches on shape, and an iterator is no worse here.
+    const SAMPLE: usize = 50;
+    let mut best = (0usize, 0usize); // (column, hits)
+    for col in 0..width {
+        let hits = rows
+            .iter()
+            .take(SAMPLE)
+            .filter(|r| r.get(col).is_some_and(|c| looks_like_phone(c)))
+            .count();
+        if hits > best.1 {
+            best = (col, hits);
+        }
+    }
+    let phone_idx = best.0;
+
+    // Name column: the first column that is not the phone column and
+    // holds mostly non-numeric text.
+    let name_idx = (0..width).find(|&col| {
+        col != phone_idx
+            && rows.iter().take(SAMPLE).any(|r| {
+                r.get(col)
+                    .is_some_and(|c| !c.is_empty() && !looks_like_phone(c))
+            })
+    });
+
+    (phone_idx, name_idx, false)
+}
+
+/// Parse CSV text into a recipient list.
+///
+/// Never returns Err for a merely messy file -- bad rows land in
+/// `skipped`. Err is reserved for "there is nothing usable here at
+/// all", which is the only case where the user must pick a different
+/// file.
+pub fn parse_recipients(input: &str) -> Result<CsvImport, String> {
+    // Strip a UTF-8 BOM; Excel writes one and it corrupts the first
+    // header cell, which would break column detection.
+    let input = input.strip_prefix('\u{feff}').unwrap_or(input);
+
+    let raw_lines: Vec<(usize, &str)> = input
+        .lines()
+        .enumerate()
+        .map(|(i, l)| (i + 1, l.trim_end_matches('\r')))
+        .filter(|(_, l)| !l.trim().is_empty())
+        .collect();
+
+    if raw_lines.is_empty() {
+        return Err("The file is empty.".into());
+    }
+
+    let rows: Vec<Vec<String>> = raw_lines.iter().map(|(_, l)| split_row(l)).collect();
+    let (phone_idx, name_idx, header) = detect_columns(&rows);
+
+    let mut out = CsvImport::default();
+    let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
+
+    for (idx, (lineno, _)) in raw_lines.iter().enumerate() {
+        if header && idx == 0 {
+            continue;
+        }
+        let row = &rows[idx];
+        let Some(cell) = row.get(phone_idx) else {
+            out.skipped.push((*lineno, "no phone column".into()));
+            continue;
+        };
+        let Some(phone) = normalise_phone(cell) else {
+            let shown = if cell.is_empty() { "(blank)" } else { cell.as_str() };
+            out.skipped
+                .push((*lineno, format!("not a phone number: {shown}")));
+            continue;
+        };
+        if !seen.insert(phone.clone()) {
+            out.duplicates += 1;
+            continue;
+        }
+        let name = name_idx
+            .and_then(|i| row.get(i))
+            .map(|s| s.to_string())
+            .unwrap_or_default();
+        out.recipients.push(CsvRecipient { phone, name });
+    }
+
+    if out.recipients.is_empty() {
+        return Err(format!(
+            "No phone numbers found. Checked {} row(s); the file needs a \
+             column of numbers such as 0712345678 or +254712345678.",
+            raw_lines.len()
+        ));
+    }
+    Ok(out)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_a_plain_two_column_file_with_header() {
+        let csv = "name,phone\nAlice,0712345678\nBob,+254798765432\n";
+        let out = parse_recipients(csv).unwrap();
+        assert_eq!(out.recipients.len(), 2);
+        assert_eq!(out.recipients[0].name, "Alice");
+        assert_eq!(out.recipients[0].phone, "+254712345678");
+        assert_eq!(out.recipients[1].phone, "+254798765432");
+        assert!(out.skipped.is_empty());
+    }
+
+    #[test]
+    fn parses_a_bare_list_with_no_header_at_all() {
+        let csv = "0712345678\n0798765432\n";
+        let out = parse_recipients(csv).unwrap();
+        assert_eq!(out.recipients.len(), 2);
+        assert_eq!(out.recipients[0].phone, "+254712345678");
+    }
+
+    /// A header cell named "phone" must not be sent a message.
+    #[test]
+    fn the_header_row_is_never_a_recipient() {
+        let csv = "name,phone\nAlice,0712345678\n";
+        let out = parse_recipients(csv).unwrap();
+        assert_eq!(out.recipients.len(), 1);
+        assert!(!out.recipients.iter().any(|r| r.name == "name"));
+    }
+
+    /// The phone column is not always second, and not always named.
+    #[test]
+    fn finds_the_phone_column_by_content_when_unnamed() {
+        let csv = "Acme Ltd,Nairobi,0712345678,acme@example.com\n\
+                   Beta Co,Mombasa,0798765432,beta@example.com\n";
+        let out = parse_recipients(csv).unwrap();
+        assert_eq!(out.recipients.len(), 2);
+        assert_eq!(out.recipients[0].phone, "+254712345678");
+    }
+
+    /// An email in the row must not be mistaken for a number.
+    #[test]
+    fn an_email_column_is_not_treated_as_phones() {
+        let csv = "phone,email\n0712345678,a@b.com\n";
+        let out = parse_recipients(csv).unwrap();
+        assert_eq!(out.recipients.len(), 1);
+        assert_eq!(out.recipients[0].phone, "+254712345678");
+    }
+
+    #[test]
+    fn quoted_fields_containing_commas_do_not_shift_columns() {
+        let csv = "name,phone\n\"Acme, Inc.\",0712345678\n";
+        let out = parse_recipients(csv).unwrap();
+        assert_eq!(out.recipients.len(), 1);
+        assert_eq!(out.recipients[0].name, "Acme, Inc.");
+        assert_eq!(out.recipients[0].phone, "+254712345678");
+    }
+
+    #[test]
+    fn semicolon_and_tab_separated_files_also_work() {
+        let out = parse_recipients("name;phone\nAlice;0712345678\n").unwrap();
+        assert_eq!(out.recipients[0].phone, "+254712345678");
+        let out = parse_recipients("name\tphone\nBob\t0798765432\n").unwrap();
+        assert_eq!(out.recipients[0].phone, "+254798765432");
+    }
+
+    /// Excel writes a BOM. It must not corrupt header detection.
+    #[test]
+    fn a_utf8_bom_does_not_break_the_header() {
+        let csv = "\u{feff}name,phone\nAlice,0712345678\n";
+        let out = parse_recipients(csv).unwrap();
+        assert_eq!(out.recipients.len(), 1);
+        assert_eq!(out.recipients[0].name, "Alice");
+    }
+
+    /// One bad row must not reject the whole file -- the directory
+    /// importer's behaviour, and the reason it is unusable here.
+    #[test]
+    fn bad_rows_are_skipped_and_reported_not_fatal() {
+        let csv = "name,phone\nAlice,0712345678\nBroken,N/A\nBob,0798765432\n";
+        let out = parse_recipients(csv).unwrap();
+        assert_eq!(out.recipients.len(), 2);
+        assert_eq!(out.skipped.len(), 1);
+        assert_eq!(out.skipped[0].0, 3, "should name the offending line");
+    }
+
+    /// Sending the same person the same marketing text twice is a bug
+    /// that costs money, so dedupe is not cosmetic.
+    #[test]
+    fn duplicate_numbers_are_collapsed_even_when_written_differently() {
+        let csv = "phone\n0712345678\n+254712345678\n254712345678\n";
+        let out = parse_recipients(csv).unwrap();
+        assert_eq!(out.recipients.len(), 1);
+        assert_eq!(out.duplicates, 2);
+    }
+
+    #[test]
+    fn an_empty_file_is_an_error_not_an_empty_batch() {
+        assert!(parse_recipients("").is_err());
+        assert!(parse_recipients("\n\n  \n").is_err());
+    }
+
+    #[test]
+    fn a_file_with_no_numbers_at_all_is_an_error() {
+        let err = parse_recipients("name,email\nAlice,a@b.com\n").unwrap_err();
+        assert!(err.contains("No phone numbers"), "got: {err}");
+    }
+
+    #[test]
+    fn normalises_the_common_kenyan_forms_identically() {
+        for raw in ["0712345678", "254712345678", "+254712345678", "712345678"] {
+            assert_eq!(
+                normalise_phone(raw).as_deref(),
+                Some("+254712345678"),
+                "failed for {raw}"
+            );
+        }
+    }
+
+    #[test]
+    fn strips_spaces_dashes_and_parens() {
+        assert_eq!(
+            normalise_phone("+254 (712) 345-678").as_deref(),
+            Some("+254712345678")
+        );
+    }
+
+    /// An international number must keep its own country code.
+    #[test]
+    fn a_non_kenyan_international_number_is_left_alone() {
+        assert_eq!(normalise_phone("+14155552671").as_deref(), Some("+14155552671"));
+        assert_eq!(normalise_phone("+442071838750").as_deref(), Some("+442071838750"));
+    }
+
+    #[test]
+    fn rejects_things_that_are_not_numbers() {
+        for raw in ["", "   ", "N/A", "none", "a@b.com", "call me", "12", "abc123"] {
+            assert!(normalise_phone(raw).is_none(), "should reject {raw:?}");
+        }
+    }
+
+    /// 16+ digits is not dialable; do not hand it to the radio.
+    #[test]
+    fn rejects_absurdly_long_numbers() {
+        assert!(normalise_phone("+1234567890123456789").is_none());
+    }
+
+    #[test]
+    fn summary_mentions_duplicates_and_skips() {
+        let csv = "phone\n0712345678\n0712345678\nbroken\n";
+        let out = parse_recipients(csv).unwrap();
+        let s = out.summary();
+        assert!(s.contains("1 recipient"), "got: {s}");
+        assert!(s.contains("duplicate"), "got: {s}");
+        assert!(s.contains("skipped"), "got: {s}");
+    }
+}

crates/apps/nigig-sms/src/sms_frame/bulk/sms_bulk_page.rs

@@ -15,6 +15,7 @@ use super::companies_list::{
 use super::directory_scraper::{
     get_extraction_state, run_extraction, DirectoryScraperAction, ExtractionState,
 };
+use super::recipient_csv::{parse_recipients, CsvImport};
 use crate::chats::sms_frame::lock_ext::LockRecover;
 use crate::chats::sms_frame::sms_utils::now_ms;
 
@@ -31,6 +32,20 @@ struct BulkSendProgress {
 /// D5: guards against two overlapping bulk sends.
 static BULK_SEND_IN_FLIGHT: AtomicBool = AtomicBool::new(false);
 static BULK_SEND_PROGRESS: Mutex<Option<BulkSendProgress>> = Mutex::new(None);
+/// Set by the UI to stop a paced batch early.
+///
+/// Pacing turns a bulk send from a few seconds into potentially hours,
+/// so "wait for it to finish" stops being an acceptable answer: the user
+/// needs a way out that does not involve killing the app. The worker
+/// checks this while sleeping, not just between sends.
+static BULK_SEND_CANCEL: AtomicBool = AtomicBool::new(false);
+
+/// Recipients parsed from a picked CSV, handed back to the UI thread.
+///
+/// The file-picker callback runs off the UI thread, so it cannot touch
+/// `Cx`. Same shape as the D1/D5 worker handoff: park the result, raise
+/// the signal, drain it in handle_event.
+static PENDING_CSV_IMPORT: Mutex<Option<Result<CsvImport, String>>> = Mutex::new(None);
 
 static FILTER_TEXT: Mutex<String> = Mutex::new(String::new());
 static SHOW_FAVORITES: Mutex<bool> = Mutex::new(false);
@@ -228,8 +243,36 @@ company_card := mod.widgets.SmsBulkCompanyCard {}
                 Label { text: "Recipients" draw_text +: { color: (LABEL), text_style.font_size: 11.0 } }
                 bulk_recipients := TextInput { width: Fill, height: 100, empty_text: "+254712345678\n+254798765432", draw_bg.color: (INPUT_BG), draw_text +: { color: (TEXT), text_style.font_size: 13.0 } }
                 bulk_recipient_count := Label { width: Fill, text: "0 recipients" draw_text +: { color: (LABEL), text_style.font_size: 10.0 } }
+
+                // Import recipients from any CSV on the device.
+                csv_import_row := View {
+                    width: Fill, height: Fit
+                    flow: Right
+                    spacing: 8
+                    csv_import_btn := Button {
+                        width: Fill, height: 34
+                        text: "📂 Import recipients from CSV"
+                        draw_bg +: { color: (ACCENT), border_radius: 8.0 }
+                        draw_text +: { color: #x000000, text_style.font_size: 12.0 }
+                    }
+                }
+                csv_import_status := Label { width: Fill, text: "" draw_text +: { color: (LABEL), text_style.font_size: 10.0 } }
+
                 Label { text: "Message" draw_text +: { color: (LABEL), text_style.font_size: 11.0 } }
                 bulk_body := TextInput { width: Fill, height: 100, empty_text: "Type your message here...", draw_bg.color: (INPUT_BG), draw_text +: { color: (TEXT), text_style.font_size: 14.0 } }
+
+                // Gap between messages. Defaults to the carrier-safe 60s
+                // rather than 0, so a long list completes instead of
+                // stopping at the throttle.
+                Label { text: "Seconds between messages" draw_text +: { color: (LABEL), text_style.font_size: 11.0 } }
+                bulk_delay_row := View {
+                    width: Fill, height: Fit
+                    flow: Right
+                    spacing: 8
+                    bulk_delay := TextInput { width: 90, height: 34, text: "60", empty_text: "60", draw_bg.color: (INPUT_BG), draw_text +: { color: (TEXT), text_style.font_size: 13.0 } }
+                    bulk_delay_hint := Label { width: Fill, text: "" draw_text +: { color: (LABEL), text_style.font_size: 10.0 } }
+                }
+
                 bulk_send_btn := Button { width: Fill, height: 48, text: "Send Bulk SMS", draw_bg +: { color: (SEND_BG), border_radius: 14.0 }, draw_text +: { color: #xFFFFFF, text_style.font_size: 15.0 } }
                 bulk_body_cost := Label { width: Fill, text: "" draw_text +: { color: (LABEL), text_style.font_size: 10.0 } }
                 bulk_status := Label { width: Fill, text: "" draw_text +: { color: (LABEL), text_style.font_size: 11.0 } }
@@ -312,6 +355,7 @@ impl Widget for SmsBulkPage {
         // D5: the bulk-send worker publishes progress and signals the UI.
         if matches!(event, Event::Signal) {
             self.drain_bulk_send_progress(cx);
+            self.drain_csv_import(cx);
         }
 
         let actions = cx.capture_actions(|cx| self.view.handle_event(cx, event, scope));
@@ -355,7 +399,27 @@ impl Widget for SmsBulkPage {
             }
         }
         if self.view.button(cx, ids!(bulk_send_btn)).clicked(&actions) {
-            self.handle_send_bulk(cx);
+            // While a paced batch is running the same button cancels it:
+            // a 200-message run at 60s apart is over three hours, and
+            // force-quitting the app is not an acceptable stop button.
+            if BULK_SEND_IN_FLIGHT.load(Ordering::Relaxed) {
+                BULK_SEND_CANCEL.store(true, Ordering::Relaxed);
+                self.view
+                    .label(cx, ids!(bulk_status))
+                    .set_text(cx, "Stopping after the current message…");
+                self.view.redraw(cx);
+            } else {
+                self.handle_send_bulk(cx);
+            }
+        }
+
+        if self.view.button(cx, ids!(csv_import_btn)).clicked(&actions) {
+            self.handle_pick_csv(cx);
+        }
+
+        // Re-estimate the batch duration as the delay is edited.
+        if self.view.text_input(cx, ids!(bulk_delay)).changed(&actions).is_some() {
+            self.refresh_delay_hint(cx);
         }
         // F10 / A5: show what this body will actually COST, live.
         //
@@ -670,22 +734,38 @@ impl SmsBulkPage {
             .as_ref()
             .is_some_and(|(b, n)| b == trimmed_body && *n == total);
 
+        let pacing = self.current_pacing(cx);
+
         if !confirmation_matches {
             self.pending_bulk_send = Some((trimmed_body.to_string(), total));
             let encoding = match seg.encoding {
                 robius_sms::SmsEncoding::Gsm7 => "GSM-7",
                 robius_sms::SmsEncoding::Ucs2 => "Unicode",
             };
+            // Quote the wall-clock duration too. At the default 60s gap
+            // a 200-recipient batch runs for over three hours, and the
+            // user must know that before the first tap, not discover it
+            // afterwards.
+            let timing = if pacing.is_immediate() {
+                String::new()
+            } else {
+                format!(
+                    " Sending {}s apart, so this takes {}.",
+                    pacing.delay_ms() / 1000,
+                    pacing.describe_duration(total)
+                )
+            };
             self.view.label(cx, ids!(bulk_status)).set_text(
                 cx,
                 &format!(
-                    "Send {} SMS ({} {} segment{} each, {} total)? \
+                    "Send {} SMS ({} {} segment{} each, {} total)?{} \
                      This cannot be undone. Tap Send again to confirm.",
                     total,
                     seg.segments,
                     encoding,
                     if seg.segments == 1 { "" } else { "s" },
                     total_segments,
+                    timing,
                 ),
             );
             self.view.redraw(cx);
@@ -733,6 +813,7 @@ impl SmsBulkPage {
         }
 
         let body_owned = trimmed_body.to_string();
+        BULK_SEND_CANCEL.store(false, Ordering::Relaxed);
         self.view
             .label(cx, ids!(bulk_status))
             .set_text(cx, &format!("Sending 0/{total}…"));
@@ -747,11 +828,51 @@ impl SmsBulkPage {
             let mut failed = 0usize;
             let mut throttled = 0usize;
 
-            for recipient in &recipients {
+            for (i, recipient) in recipients.iter().enumerate() {
-                if limiter.allow_at(now_ms()).is_err() {
+                // Wait BEFORE every send except the first. Doing it here
+                // rather than at the end of the body means a cancel
+                // between messages takes effect without burning the gap,
+                // and the arithmetic matches SendPacing::total_duration_ms
+                // (n messages, n-1 gaps).
+                if i > 0 && !pacing.is_immediate() {
+                    let step = std::time::Duration::from_millis(200);
+                    let mut waited = 0i64;
+                    while waited < pacing.delay_ms() {
+                        if BULK_SEND_CANCEL.load(Ordering::Relaxed) {
+                            break;
+                        }
+                        thread::sleep(step);
+                        waited += step.as_millis() as i64;
+                    }
+                }
+
+                if BULK_SEND_CANCEL.load(Ordering::Relaxed) {
                     throttled = recipients.len() - sent - failed;
                     break;
                 }
+
+                // The limiter is still the backstop. With a sane gap it
+                // never fires; if the user forced delay=0 on a list
+                // longer than the cap, wait it out rather than
+                // abandoning the batch the way this loop used to.
+                loop {
+                    match limiter.allow_at(now_ms()) {
+                        Ok(()) => break,
+                        Err(wait_ms) => {
+                            if BULK_SEND_CANCEL.load(Ordering::Relaxed) {
+                                break;
+                            }
+                            // Cap each nap so cancel stays responsive.
+                            let nap = wait_ms.clamp(0, 1000) as u64;
+                            thread::sleep(std::time::Duration::from_millis(nap.max(50)));
+                        }
+                    }
+                }
+                if BULK_SEND_CANCEL.load(Ordering::Relaxed) {
+                    throttled = recipients.len() - sent - failed;
+                    break;
+                }
+
                 match send_single(recipient, &body_owned) {
                     Ok(()) => sent += 1,
                     Err(_) => failed += 1,
@@ -775,10 +896,146 @@ impl SmsBulkPage {
                 done: true,
             });
             BULK_SEND_IN_FLIGHT.store(false, Ordering::Relaxed);
+            BULK_SEND_CANCEL.store(false, Ordering::Relaxed);
             SignalToUI::set_ui_signal();
         });
     }
 
+    /// Current pacing from the delay field.
+    ///
+    /// Empty or unparseable falls back to the carrier-safe default
+    /// rather than to zero: a typo must not turn a paced batch into a
+    /// burst that trips the throttle at message 30.
+    fn current_pacing(&mut self, cx: &mut Cx) -> robius_sms::SendPacing {
+        let raw = self.view.text_input(cx, ids!(bulk_delay)).text();
+        let t = raw.trim();
+        if t.is_empty() {
+            return robius_sms::SendPacing::default();
+        }
+        match t.parse::<i64>() {
+            Ok(secs) => robius_sms::SendPacing::from_seconds(secs),
+            Err(_) => robius_sms::SendPacing::default(),
+        }
+    }
+
+    /// Show what the chosen delay means for this recipient count.
+    fn refresh_delay_hint(&mut self, cx: &mut Cx) {
+        let n = self
+            .view
+            .text_input(cx, ids!(bulk_recipients))
+            .text()
+            .lines()
+            .filter(|l| !l.trim().is_empty())
+            .count();
+        let pacing = self.current_pacing(cx);
+        let hint = if n == 0 {
+            String::new()
+        } else if pacing.is_immediate() {
+            format!(
+                "No delay — sends as fast as allowed, and stops at the \
+                 carrier limit of {} per {} min.",
+                robius_sms::SendRateLimiter::DEFAULT_CAPACITY,
+                robius_sms::SendRateLimiter::DEFAULT_WINDOW_MS / 60_000,
+            )
+        } else {
+            format!(
+                "{} message(s), {}s apart — {}.",
+                n,
+                pacing.delay_ms() / 1000,
+                pacing.describe_duration(n)
+            )
+        };
+        self.view.label(cx, ids!(bulk_delay_hint)).set_text(cx, &hint);
+        self.view.redraw(cx);
+    }
+
+    /// Open the system file picker and parse whatever CSV comes back.
+    ///
+    /// Deliberately not `directory_scraper::find_csv`, which only looks
+    /// for one hardcoded filename in /sdcard/Download and tells the user
+    /// to run `adb push` when it is missing. This accepts any file the
+    /// user can reach through the platform picker, including Drive and
+    /// other storage providers, which is what "upload a CSV" means on a
+    /// phone.
+    fn handle_pick_csv(&mut self, cx: &mut Cx) {
+        self.view
+            .label(cx, ids!(csv_import_status))
+            .set_text(cx, "Opening file picker…");
+        self.view.redraw(cx);
+
+        use robius_file_picker::FileDialog;
+        let _ = FileDialog::new()
+            .add_filter("CSV", &["csv", "txt"])
+            .set_title("Pick a CSV of recipients")
+            .pick_file(|result| {
+                // Runs off the UI thread: park the outcome and signal.
+                let parked = match result {
+                    Ok(Some(picked)) => match picked.into_local_file() {
+                        Ok(local) => match std::fs::read_to_string(local.path()) {
+                            Ok(text) => Some(parse_recipients(&text)),
+                            Err(e) => Some(Err(format!("Could not read the file: {e}"))),
+                        },
+                        Err(e) => Some(Err(format!("Could not open the file: {e}"))),
+                    },
+                    // User cancelled: leave the previous list alone.
+                    Ok(None) => None,
+                    Err(e) => Some(Err(format!("File picker failed: {e}"))),
+                };
+                if let Some(outcome) = parked {
+                    *PENDING_CSV_IMPORT.lock_recover() = Some(outcome);
+                    SignalToUI::set_ui_signal();
+                }
+            });
+    }
+
+    /// Apply a CSV import parked by the picker callback, if any.
+    fn drain_csv_import(&mut self, cx: &mut Cx) {
+        let Some(outcome) = PENDING_CSV_IMPORT.lock_recover().take() else {
+            return;
+        };
+        match outcome {
+            Err(e) => {
+                self.view.label(cx, ids!(csv_import_status)).set_text(cx, &e);
+            }
+            Ok(import) => {
+                let numbers: Vec<String> =
+                    import.recipients.iter().map(|r| r.phone.clone()).collect();
+                self.view
+                    .text_input(cx, ids!(bulk_recipients))
+                    .set_text(cx, &numbers.join("\n"));
+                self.view
+                    .label(cx, ids!(bulk_recipient_count))
+                    .set_text(cx, &format!("{} recipients", numbers.len()));
+
+                // Name the first few skipped lines. "3 rows skipped" with
+                // no detail is not actionable when the file has 900 rows.
+                let mut status = format!("Imported {}", import.summary());
+                if !import.skipped.is_empty() {
+                    let detail: Vec<String> = import
+                        .skipped
+                        .iter()
+                        .take(3)
+                        .map(|(line, why)| format!("line {line}: {why}"))
+                        .collect();
+                    status.push_str(&format!(" — {}", detail.join("; ")));
+                    if import.skipped.len() > 3 {
+                        status.push_str(&format!(" (+{} more)", import.skipped.len() - 3));
+                    }
+                }
+                self.view
+                    .label(cx, ids!(csv_import_status))
+                    .set_text(cx, &status);
+
+                // A freshly imported list invalidates any armed
+                // confirmation: the count it was armed for is stale.
+                self.pending_bulk_send = None;
+                self.set_sub_tab(cx, BulkSubTab::Compose);
+                self.refresh_delay_hint(cx);
+            }
+        }
+        self.view.redraw(cx);
+    }
+
     /// Apply bulk-send progress published by the D5 worker, if any.
     fn drain_bulk_send_progress(&mut self, cx: &mut Cx) {
         let Some(p) = BULK_SEND_PROGRESS.lock_recover().take() else {

crates/robius-sms/src/lib.rs

@@ -335,6 +335,115 @@ impl Default for SendRateLimiter {
     }
 }
 
+/// How long to wait between two sends in a batch, and how long a whole
+/// batch will therefore take.
+///
+/// `SendRateLimiter` above answers "may I send now?" and, when the
+/// answer is no, the bulk worker used to `break` -- abandoning the rest
+/// of the list and telling the user to come back later. That is correct
+/// for protecting the carrier ceiling but useless as a way to actually
+/// deliver 200 messages: the user has to babysit the app and re-run it
+/// seven times.
+///
+/// Pacing is the other half. Instead of firing as fast as the loop
+/// allows and then stopping dead at the cap, put a fixed gap between
+/// sends so the batch stays under the ceiling by construction and runs
+/// to completion.
+///
+/// Pure arithmetic, no sleeping, so the schedule can be asserted in a
+/// unit test on any host.
+#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+pub struct SendPacing {
+    /// Gap between consecutive sends. Zero means "as fast as possible",
+    /// which is still bounded by `SendRateLimiter`.
+    delay_ms: i64,
+}
+
+impl SendPacing {
+    /// Longest gap we let a user choose: 10 minutes. Past this a batch
+    /// of any size takes days, and the app is not a scheduler.
+    pub const MAX_DELAY_MS: i64 = 10 * 60 * 1000;
+
+    /// Gap that keeps a batch just inside the default carrier ceiling
+    /// (30 messages / 30 minutes), i.e. one per minute.
+    pub const RECOMMENDED_DELAY_MS: i64 =
+        SendRateLimiter::DEFAULT_WINDOW_MS / SendRateLimiter::DEFAULT_CAPACITY as i64;
+
+    /// Clamps rather than rejecting: this is driven by a text field, and
+    /// refusing to send because someone typed 9999 would be worse than
+    /// quietly using the maximum.
+    pub fn from_millis(delay_ms: i64) -> Self {
+        Self {
+            delay_ms: delay_ms.clamp(0, Self::MAX_DELAY_MS),
+        }
+    }
+
+    pub fn from_seconds(delay_s: i64) -> Self {
+        Self::from_millis(delay_s.saturating_mul(1000))
+    }
+
+    /// A gap that spreads `count` messages evenly across the rate
+    /// limiter's window, so the batch never trips the cap.
+    ///
+    /// Returns `RECOMMENDED_DELAY_MS` when the batch already fits.
+    pub fn to_stay_under(capacity: u32, window_ms: i64, count: usize) -> Self {
+        if capacity == 0 || count == 0 {
+            return Self::from_millis(0);
+        }
+        if count <= capacity as usize {
+            return Self::from_millis(0);
+        }
+        // Need `count` sends; only `capacity` fit per window.
+        Self::from_millis(window_ms / capacity as i64)
+    }
+
+    pub fn delay_ms(self) -> i64 {
+        self.delay_ms
+    }
+
+    pub fn is_immediate(self) -> bool {
+        self.delay_ms == 0
+    }
+
+    /// Wall-clock duration of a batch of `count` messages.
+    ///
+    /// There are `count - 1` gaps, not `count`: nothing is waited before
+    /// the first message or after the last. Off-by-one here would show
+    /// the user a materially wrong estimate on small batches.
+    pub fn total_duration_ms(self, count: usize) -> i64 {
+        let gaps = count.saturating_sub(1) as i64;
+        self.delay_ms.saturating_mul(gaps)
+    }
+
+    /// Human estimate for the confirmation prompt, e.g. "about 3m 20s".
+    pub fn describe_duration(self, count: usize) -> String {
+        let ms = self.total_duration_ms(count);
+        if ms <= 0 {
+            return "a few seconds".into();
+        }
+        let total_s = ms / 1000;
+        let h = total_s / 3600;
+        let m = (total_s % 3600) / 60;
+        let s = total_s % 60;
+        if h > 0 {
+            format!("about {h}h {m}m")
+        } else if m > 0 {
+            format!("about {m}m {s}s")
+        } else {
+            format!("about {s}s")
+        }
+    }
+}
+
+impl Default for SendPacing {
+    /// Default to the carrier-safe gap rather than to zero. A user who
+    /// never touches the field gets a batch that completes instead of
+    /// one that stops a third of the way through.
+    fn default() -> Self {
+        Self::from_millis(Self::RECOMMENDED_DELAY_MS)
+    }
+}
+
 impl BulkSendRequest {
     /// Validate a bulk request.
     ///
@@ -864,6 +973,109 @@ mod tests {
         assert_eq!(rl.in_window(31), 30);
     }
 
+    // ---- SendPacing -----------------------------------------------------
+
+    #[test]
+    fn pacing_defaults_to_the_carrier_safe_gap_not_to_zero() {
+        // A user who never touches the delay field must still get a
+        // batch that completes rather than one that stops at 30.
+        let p = SendPacing::default();
+        assert_eq!(p.delay_ms(), SendPacing::RECOMMENDED_DELAY_MS);
+        assert!(!p.is_immediate());
+    }
+
+    #[test]
+    fn the_recommended_gap_is_one_per_minute() {
+        // 30 minutes / 30 messages.
+        assert_eq!(SendPacing::RECOMMENDED_DELAY_MS, 60_000);
+    }
+
+    /// The point of pacing: a batch bigger than the cap must still
+    /// finish. Walk the limiter with the recommended gap and check
+    /// nothing is refused.
+    #[test]
+    fn the_recommended_gap_keeps_a_long_batch_under_the_limiter() {
+        let p = SendPacing::default();
+        let mut rl = SendRateLimiter::default();
+        let mut now = 0i64;
+        for i in 0..200 {
+            assert!(
+                rl.allow_at(now).is_ok(),
+                "send {i} refused at t={now}ms; pacing failed to protect the cap"
+            );
+            now += p.delay_ms();
+        }
+    }
+
+    /// And the negative: with no gap, the same batch dies at 30. This is
+    /// the behaviour that made the old bulk worker abandon the list.
+    #[test]
+    fn without_pacing_the_same_batch_is_refused_at_the_cap() {
+        let mut rl = SendRateLimiter::default();
+        let allowed = (0..200).filter(|i| rl.allow_at(*i).is_ok()).count();
+        assert_eq!(allowed, SendRateLimiter::DEFAULT_CAPACITY as usize);
+    }
+
+    #[test]
+    fn pacing_clamps_instead_of_rejecting_out_of_range_input() {
+        assert_eq!(SendPacing::from_millis(-5).delay_ms(), 0);
+        assert_eq!(
+            SendPacing::from_millis(i64::MAX).delay_ms(),
+            SendPacing::MAX_DELAY_MS
+        );
+        assert_eq!(
+            SendPacing::from_seconds(i64::MAX).delay_ms(),
+            SendPacing::MAX_DELAY_MS,
+            "seconds->millis must not overflow into a negative"
+        );
+    }
+
+    /// n messages have n-1 gaps. Getting this wrong shows the user a
+    /// visibly wrong estimate on a 2-message batch.
+    #[test]
+    fn duration_counts_the_gaps_between_messages_not_the_messages() {
+        let p = SendPacing::from_seconds(10);
+        assert_eq!(p.total_duration_ms(0), 0);
+        assert_eq!(p.total_duration_ms(1), 0, "one message waits for nothing");
+        assert_eq!(p.total_duration_ms(2), 10_000);
+        assert_eq!(p.total_duration_ms(11), 100_000);
+    }
+
+    #[test]
+    fn duration_description_is_human_and_never_empty() {
+        assert_eq!(SendPacing::from_seconds(0).describe_duration(50), "a few seconds");
+        assert_eq!(SendPacing::from_seconds(10).describe_duration(4), "about 30s");
+        assert_eq!(SendPacing::from_seconds(60).describe_duration(4), "about 3m 0s");
+        assert!(SendPacing::from_seconds(60)
+            .describe_duration(200)
+            .starts_with("about 3h"));
+    }
+
+    #[test]
+    fn to_stay_under_is_free_when_the_batch_already_fits() {
+        let p = SendPacing::to_stay_under(30, 30 * 60 * 1000, 10);
+        assert!(p.is_immediate(), "10 messages fit in a 30 cap; no need to wait");
+    }
+
+    #[test]
+    fn to_stay_under_spreads_a_batch_that_does_not_fit() {
+        let p = SendPacing::to_stay_under(30, 30 * 60 * 1000, 200);
+        assert_eq!(p.delay_ms(), 60_000);
+
+        let mut rl = SendRateLimiter::new(30, 30 * 60 * 1000);
+        let mut now = 0i64;
+        for _ in 0..200 {
+            assert!(rl.allow_at(now).is_ok());
+            now += p.delay_ms();
+        }
+    }
+
+    #[test]
+    fn to_stay_under_handles_degenerate_input_without_panicking() {
+        assert!(SendPacing::to_stay_under(0, 1000, 10).is_immediate());
+        assert!(SendPacing::to_stay_under(30, 1000, 0).is_immediate());
+    }
+
     #[test]
     fn a_bulk_blast_of_200_is_stopped_well_short() {
         // The scenario the bulk UI makes one tap away.