diff --git a/AGENTS.md b/AGENTS.md
index 528976f..fb8b7ef 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -15,7 +15,7 @@ bun run lint      # run oxlint
 bun run fmt       # format with oxfmt (fmt:check to verify)
 ```
 
-There are no tests in this project.
+Run `bun test tests/purge-command.test.js` for the mocked bulk-purge regression tests (no Slack or database access required).
 
 ## Architecture
 
@@ -60,6 +60,10 @@ when it already matches the home workspace.
 
 **Rate limiter** (`lib/ratelimiter.js`): Handles Slack API rate limits with exponential backoff. Used by `lib/purge.js` for batch message deletion.
 
+**Bulk purge** (`lib/commands/purge.js`): `/pro purge <count> [true|false]` is the Discord-`?purge`-style bulk delete — it grabs the last `<count>` (max 100) top-level messages from `conversations.history` (thread replies aren't returned by that endpoint, so they're excluded from the count automatically) and confirms via a modal before deleting anything. The trailing `true`/`false` argument controls thread replies: `false` (default) deletes only the matched top-level messages, same batch-delete path (`RateLimiter.deleteBatch`) as before, logged via dedicated `logPurge`/`notifyPurgeDeletion`/`publicLogPurge` exports (parallel to `logThread`/`notifyThreadDeletion`/`publicLogThread` but worded for a purge instead of a single thread). `true` instead routes any matched message that has replies through `purge()` from `lib/purge.js` — the same full thread-destroy path as the Destroy Thread shortcut, with its own retry loop and logging — so parent and replies are removed together instead of leaving an orphaned thread behind. Because deletion only ever considers messages `conversations.history` can currently see, running `true` again later does not retroactively clean up threads whose parent was already removed by an earlier `false` run — that orphaned parent ts is no longer a message history can return.
+
+The confirmation modal also takes two independent ways to protect messages, both parsed by `parseTimestampToken` from either a raw Slack ts or a permalink's `p<16 digits>` suffix: a "keep these messages" list (individually excluded messages never count toward `<count>` — history keeps paging back to make up the difference), and a "keep everything between two messages" range (exactly two links define an inclusive span; any other count of links is treated as a mistake and aborts the purge with nothing deleted, rather than guessing). A checkbox controls whether the kept range counts toward `<count>`: off (default) behaves like the individual exclusions and digs further back; on treats `<count>` as a fixed window so protected messages reduce how many actually get deleted. History and reply scans use the user token. With thread deletion enabled, all reply pages are checked before selecting a thread; if any reply is excluded or inside the kept range, the whole thread is skipped without consuming the count.
+
 **Web dashboard and API** (`lib/web/`): A Hono app served by `lib/web/server.js`, mounted by
 `index.js` alongside the bot and also runnable on its own with `bun run start:web`. Sign-in is
 Slack OAuth through Better Auth (`auth.js`), restricted to the workspace the bot is installed in.
diff --git a/lib/commands/help.js b/lib/commands/help.js
index ea2385e..dee631c 100644
--- a/lib/commands/help.js
+++ b/lib/commands/help.js
@@ -17,6 +17,10 @@ const MODERATOR_CMDS = [
 ];
 
 const MANAGER_CMDS = [
+  {
+    cmd: "/pro purge <count> [true|false]",
+    desc: "Bulk-delete the last <count> top-level messages; `true` also nukes their threads",
+  },
   { cmd: "/pro embeds", desc: "Manage blacklisted embeds" },
   { cmd: "/pro welcome [set|remove|view]", desc: "Manage the welcome message for new peeps!" },
   { cmd: "/pro gate set [button|phrase]", desc: "Require an acknowledgement before posting" },
diff --git a/lib/commands/purge.js b/lib/commands/purge.js
new file mode 100644
index 0000000..b1910ce
--- /dev/null
+++ b/lib/commands/purge.js
@@ -0,0 +1,407 @@
+import { canManage } from "../perms.js";
+import { RateLimiter } from "../ratelimiter.js";
+import { logPurge, notifyPurgeDeletion } from "../logger.js";
+import { publicLogPurge } from "../public-logger.js";
+import { purge as purgeThread } from "../purge.js";
+
+const rateLimiter = new RateLimiter(1000, 5);
+
+const MAX_PURGE = 100;
+const HISTORY_PAGE_SIZE = 200;
+const MAX_HISTORY_PAGES = 5;
+
+const TRUE_VALUES = new Set(["true", "yes", "y", "1"]);
+const FALSE_VALUES = new Set(["false", "no", "n", "0"]);
+
+const SKIP_SUBTYPES = new Set([
+  "channel_archive",
+  "channel_join",
+  "channel_leave",
+  "channel_name",
+  "channel_purpose",
+  "channel_topic",
+  "channel_unarchive",
+  "group_join",
+  "group_leave",
+  "group_name",
+  "group_purpose",
+  "group_topic",
+  "group_unarchive",
+  "message_deleted",
+  "message_changed",
+]);
+
+const eph = (text) => ({ response_type: "ephemeral", text });
+const usage = () =>
+  eph(
+    `Usage: \`/pro purge <count> [true|false]\` — count is 1-${MAX_PURGE} top-level messages. Add \`true\` to also delete each matched message's thread replies (default \`false\`, replies are left alone).`,
+  );
+
+function isThreaded(message) {
+  return Boolean(message.reply_count) || (message.thread_ts && message.thread_ts === message.ts);
+}
+
+function isPurgeable(message) {
+  return Boolean(message?.ts) && !SKIP_SUBTYPES.has(message.subtype);
+}
+
+function inRange(ts, range) {
+  if (!range) return false;
+  const t = parseFloat(ts);
+  return t >= range.start && t <= range.end;
+}
+
+// Message permalinks look like .../archives/C123/p1690000000123456 — the 16-digit
+// suffix is the ts with the decimal point removed (10s + 6 micros).
+function parseTimestampToken(token) {
+  const bare = token.match(/^\d{10}\.\d{6}$/);
+  if (bare) return token;
+
+  const link = token.match(/\/p(\d{10})(\d{6})(?:[/?].*)?$/);
+  if (link) return `${link[1]}.${link[2]}`;
+
+  return null;
+}
+
+function parseExcludedTimestamps(raw) {
+  const excluded = new Set();
+  if (!raw) return excluded;
+
+  for (const line of raw.split(/\r?\n/)) {
+    const token = line.trim();
+    if (!token) continue;
+    const ts = parseTimestampToken(token);
+    if (ts) excluded.add(ts);
+  }
+
+  return excluded;
+}
+
+// Exactly two links/timestamps define an inclusive range to keep. Anything else
+// (one link, three links, unparseable text) is a mistake we surface rather than
+// guess at — silently keeping the wrong span would be worse than refusing.
+function parseRange(raw) {
+  if (!raw || !raw.trim()) return { range: null, error: null };
+
+  const parsed = raw
+    .split(/\r?\n/)
+    .map((line) => line.trim())
+    .filter(Boolean)
+    .map(parseTimestampToken)
+    .filter(Boolean);
+
+  if (parsed.length !== 2) {
+    return {
+      range: null,
+      error: `"Keep everything between two messages" needs exactly two valid message links or timestamps (found ${parsed.length}).`,
+    };
+  }
+
+  const [start, end] = parsed.map(parseFloat).sort((a, b) => a - b);
+  return { range: { start, end }, error: null };
+}
+
+async function hasProtectedReply(client, channel, threadTs, { excluded, range }) {
+  if (!excluded.size && !range) return false;
+
+  let cursor;
+  do {
+    const result = await client.conversations.replies({
+      channel,
+      ts: threadTs,
+      limit: HISTORY_PAGE_SIZE,
+      cursor,
+    });
+    if (
+      (result.messages || []).some(
+        (message) => excluded.has(message.ts) || inRange(message.ts, range),
+      )
+    ) {
+      return true;
+    }
+    cursor = result.response_metadata?.next_cursor;
+  } while (cursor);
+
+  return false;
+}
+
+// conversations.history only returns top-level channel messages; thread replies
+// are excluded automatically unless they were broadcast to the channel.
+// Individually-excluded messages never count toward `count` — history keeps
+// paging until enough deletable messages are found (or pages run out). Messages
+// kept by the range only consume a `count` slot when countRangeInTotal is set;
+// otherwise they're skipped as if they were never there, same as an exclusion.
+async function collectMessages(
+  client,
+  channel,
+  count,
+  { excluded, range, countRangeInTotal, includeThreads },
+) {
+  const toDelete = [];
+  let budgetUsed = 0;
+  let cursor;
+
+  for (let page = 0; page < MAX_HISTORY_PAGES && budgetUsed < count; page++) {
+    const result = await client.conversations.history({
+      channel,
+      limit: HISTORY_PAGE_SIZE,
+      cursor,
+    });
+
+    for (const message of result.messages || []) {
+      if (budgetUsed >= count) break;
+      if (!isPurgeable(message)) continue;
+      if (excluded.has(message.ts)) continue;
+
+      if (inRange(message.ts, range)) {
+        if (countRangeInTotal) budgetUsed++;
+        continue;
+      }
+
+      // Keep the whole thread intact if cascading would remove a protected reply.
+      // Skipped threads don't consume the budget, so continue looking further back.
+      if (
+        includeThreads &&
+        isThreaded(message) &&
+        (await hasProtectedReply(client, channel, message.ts, { excluded, range }))
+      ) {
+        continue;
+      }
+
+      toDelete.push(message);
+      budgetUsed++;
+    }
+
+    cursor = result.response_metadata?.next_cursor;
+    if (!cursor) break;
+  }
+
+  return toDelete;
+}
+
+function purgeModal(channelId, count, includeThreads) {
+  const threadNote = includeThreads
+    ? "Thread replies on any matched message will *also* be permanently deleted. Threads containing a kept reply are skipped entirely and don't count toward the total."
+    : "Thread replies are left alone — a matched message with replies still gets deleted, its thread left dangling underneath.";
+
+  return {
+    type: "modal",
+    callback_id: "purge_confirm",
+    private_metadata: JSON.stringify({ channel: channelId, count, includeThreads }),
+    title: { type: "plain_text", text: "Confirm purge" },
+    submit: { type: "plain_text", text: "Delete" },
+    close: { type: "plain_text", text: "Cancel" },
+    blocks: [
+      {
+        type: "section",
+        text: {
+          type: "mrkdwn",
+          text: `*Are you sure?* This will permanently delete up to ${count} of the most recent messages in <#${channelId}>. ${threadNote} This cannot be undone.`,
+        },
+      },
+      {
+        type: "input",
+        block_id: "reason",
+        optional: false,
+        label: { type: "plain_text", text: "Reason" },
+        element: {
+          type: "plain_text_input",
+          action_id: "reason_input",
+          multiline: true,
+          placeholder: { type: "plain_text", text: "Why are you purging these messages?" },
+        },
+        hint: { type: "plain_text", text: "This will be recorded in the audit log." },
+      },
+      {
+        type: "input",
+        block_id: "exclude",
+        optional: true,
+        label: { type: "plain_text", text: "Keep these messages" },
+        element: {
+          type: "plain_text_input",
+          action_id: "exclude_input",
+          multiline: true,
+          placeholder: { type: "plain_text", text: "Paste message links, one per line" },
+        },
+        hint: {
+          type: "plain_text",
+          text: "Optional. Right-click a message → Copy link. These are skipped entirely and don't count toward the total.",
+        },
+      },
+      {
+        type: "input",
+        block_id: "range",
+        optional: true,
+        label: { type: "plain_text", text: "Keep everything between two messages" },
+        element: {
+          type: "plain_text_input",
+          action_id: "range_input",
+          multiline: true,
+          placeholder: { type: "plain_text", text: "Paste exactly two message links" },
+        },
+        hint: {
+          type: "plain_text",
+          text: "Optional. Both messages and everything between them (in either order) are kept.",
+        },
+      },
+      {
+        type: "input",
+        block_id: "range_count",
+        optional: true,
+        label: { type: "plain_text", text: "Kept range and the total" },
+        element: {
+          type: "checkboxes",
+          action_id: "range_count_checkbox",
+          options: [
+            {
+              text: { type: "plain_text", text: "Count the kept range toward <count>" },
+              description: {
+                type: "plain_text",
+                text: "Off (default): purge digs further back to still delete <count> messages.",
+              },
+              value: "count",
+            },
+          ],
+        },
+      },
+    ],
+  };
+}
+
+async function handleView({ view, body, client, context, logger }) {
+  const { channel, count, includeThreads } = JSON.parse(view.private_metadata);
+  const userId = body.user.id;
+
+  if (!(await canManage(context.userClient, userId, channel))) {
+    logger.warn(`${userId} denied for purge_confirm`);
+    return;
+  }
+
+  const reason = view.state?.values?.reason?.reason_input?.value?.trim() || "";
+  const excludeRaw = view.state?.values?.exclude?.exclude_input?.value || "";
+  const rangeRaw = view.state?.values?.range?.range_input?.value || "";
+  const countRangeInTotal = (
+    view.state?.values?.range_count?.range_count_checkbox?.selected_options || []
+  ).some((o) => o.value === "count");
+
+  const excluded = parseExcludedTimestamps(excludeRaw);
+  const { range, error: rangeError } = parseRange(rangeRaw);
+
+  if (rangeError) {
+    await client.chat
+      .postEphemeral({
+        channel,
+        user: userId,
+        text: `:red-x: ${rangeError} Purge cancelled — nothing was deleted.`,
+      })
+      .catch((error) => logger.warn(`purge range error notice failed: ${error.message}`));
+    return;
+  }
+
+  const messages = await collectMessages(context.userClient, channel, count, {
+    excluded,
+    range,
+    countRangeInTotal,
+    includeThreads,
+  });
+  if (!messages.length) {
+    await client.chat
+      .postEphemeral({ channel, user: userId, text: "No messages found to purge." })
+      .catch((error) => logger.warn(`purge empty notice failed: ${error.message}`));
+    return;
+  }
+
+  // With includeThreads, messages with replies get destroyed via the same
+  // full-thread purge used by the Destroy Thread shortcut (parent + all
+  // replies, with its own retry loop and logging). Everything else — and
+  // every threaded message when includeThreads is off — is a plain batch
+  // delete of just the top-level message.
+  const toCascade = includeThreads ? messages.filter(isThreaded) : [];
+  const toBatchDelete = includeThreads ? messages.filter((m) => !isThreaded(m)) : messages;
+
+  let dc = 0;
+  let ec = 0;
+
+  if (toBatchDelete.length) {
+    await Promise.all([
+      logPurge(client, logger, { channel, messages: toBatchDelete, deletedBy: userId, reason }),
+      publicLogPurge(client, { channel, count: toBatchDelete.length, deletedBy: userId }),
+    ]);
+
+    const result = await rateLimiter.deleteBatch(
+      context.userClient,
+      logger,
+      channel,
+      toBatchDelete,
+      5,
+      2000,
+    );
+    dc += result.dc;
+    ec += result.ec;
+
+    await notifyPurgeDeletion(client, {
+      channel,
+      messages: toBatchDelete,
+      deletedBy: userId,
+      reason,
+    }).catch((error) => logger.warn(`purge notify failed: ${error.message}`));
+  }
+
+  for (const message of toCascade) {
+    try {
+      await purgeThread(context.userClient, logger, channel, message.ts, userId, {
+        reason,
+        notificationClient: client,
+      });
+      dc++;
+    } catch (error) {
+      ec++;
+      logger.error(`purge: failed to destroy thread ${message.ts} in ${channel}: ${error.message}`);
+    }
+  }
+
+  await client.chat
+    .postEphemeral({
+      channel,
+      user: userId,
+      text: `:okay-1: Purged ${dc} top-level message${dc === 1 ? "" : "s"}${toCascade.length ? `, including full deletion of ${toCascade.length} thread${toCascade.length === 1 ? "" : "s"}` : ""}${ec ? `, ${ec} failed` : ""}.`,
+    })
+    .catch((error) => logger.warn(`purge confirmation failed: ${error.message}`));
+
+  logger.info(
+    `purge done channel=${channel} count=${dc} errors=${ec} threads=${toCascade.length} by=${userId}`,
+  );
+}
+
+export const views = [{ callbackId: "purge_confirm", handleView }];
+
+export default {
+  name: "purge",
+  description: "Bulk-delete recent top-level messages in this channel",
+  async execute({ command, args, respond, client, context }) {
+    const channelId = command.channel_id;
+    const userId = command.user_id;
+
+    if (!(await canManage(context.userClient, userId, channelId))) {
+      return respond(eph(":loll: You do not have permission to purge this channel."));
+    }
+
+    const count = Number(args[0]);
+    if (!Number.isInteger(count) || count < 1 || count > MAX_PURGE) {
+      return respond(usage());
+    }
+
+    let includeThreads = false;
+    if (args[1] !== undefined) {
+      const flag = args[1].toLowerCase();
+      if (TRUE_VALUES.has(flag)) includeThreads = true;
+      else if (FALSE_VALUES.has(flag)) includeThreads = false;
+      else return respond(usage());
+    }
+
+    await client.views.open({
+      trigger_id: command.trigger_id,
+      view: purgeModal(channelId, count, includeThreads),
+    });
+  },
+};
diff --git a/lib/logger.js b/lib/logger.js
index 254ee35..9826a89 100644
--- a/lib/logger.js
+++ b/lib/logger.js
@@ -122,6 +122,34 @@ export async function notifyThreadDeletion(client, { channel, messages, deletedB
   );
 }
 
+export async function notifyPurgeDeletion(client, { channel, messages, deletedBy, reason }) {
+  const messagesByUser = new Map();
+  for (const message of messages) {
+    if (!message.user || message.user === deletedBy) continue;
+    const userMessages = messagesByUser.get(message.user) || [];
+    userMessages.push(message);
+    messagesByUser.set(message.user, userMessages);
+  }
+
+  await Promise.all(
+    [...messagesByUser].map(async ([user, userMessages]) => {
+      const plural = userMessages.length !== 1;
+      const quotedMessages = userMessages
+        .map((message) => quote(message.text || "_no text content_"))
+        .join("\n\n");
+      const text = [
+        `Hello, ${plural ? `${userMessages.length} of your messages` : "your message"} in <#${channel}> ${plural ? "were" : "was"} removed as part of a channel purge. ${plural ? "Here are your messages:" : "Here is your message:"}`,
+        quotedMessages.slice(0, 35000),
+        `Why was it purged?`,
+        quote(reason),
+        `This purge was run by one of the stewards for <#${channel}>. Some channels have special rules. Make sure to always read the channel topic, and ask questions if you're not sure!`,
+      ].join("\n");
+
+      await sendDirectMessage(client, user, text);
+    }),
+  );
+}
+
 export async function logBan(client, { channel, user, bannedBy, reason, expires, unbanned, api }) {
   if (!LOG_CHANNEL) return;
 
@@ -291,3 +319,88 @@ export async function logThread(
     blocks,
   });
 }
+
+export async function logPurge(client, logger, { channel, messages, deletedBy, reason, api }) {
+  if (!LOG_CHANNEL) return;
+
+  let cdnUrl = null;
+
+  const lc = getLogClient(client);
+  const names = {};
+  await Promise.all(
+    [...new Set(messages.map((m) => m.user).filter(Boolean))].map(async (u) => {
+      try {
+        const { user: info } = await lc.users.info({ user: u });
+        names[u] = info?.profile?.display_name || info?.real_name || info?.name || u;
+      } catch (err) {
+        logger.error(`failed to find user ${u}: ${err.message}`);
+        try {
+          const { user: info } = await client.users.info({ user: u });
+          names[u] = info?.profile?.display_name || info?.real_name || info?.name || u;
+        } catch {
+          names[u] = u;
+        }
+      }
+    }),
+  );
+
+  const lines = messages.map((m) => {
+    const u = m.user || "unknown";
+    return `[${new Date(parseFloat(m.ts) * 1000).toISOString()}] <${u}|${names[u] || u}> ${m.text || ""}`;
+  });
+
+  const content = lines.join("\n");
+
+  if (CDN_KEY) {
+    try {
+      const blob = new Blob([content], { type: "text/plain" });
+      const formData = new FormData();
+      formData.append("file", blob, `purge-${channel}-${Date.now()}.txt`);
+
+      const res = await fetch("https://cdn.hackclub.com/api/v4/upload", {
+        method: "POST",
+        headers: { Authorization: `Bearer ${CDN_KEY}` },
+        body: formData,
+      });
+
+      if (res.ok) {
+        const data = await res.json();
+        cdnUrl = data.url;
+      } else {
+        logger.error(`CDN upload failed: ${res.status} ${await res.text()}`);
+      }
+    } catch (err) {
+      logger.error(`CDN upload error: ${err.message}`);
+    }
+  }
+
+  const cdnLine = cdnUrl ? `*Archive:* ${cdnUrl}` : "_CDN upload skipped or failed_";
+
+  const blocks = [
+    {
+      type: "section",
+      text: {
+        type: "mrkdwn",
+        text: `:broom: <@${deletedBy}> purged ${messages.length} message${messages.length === 1 ? "" : "s"} in <#${channel}>.${apiTag(api)}`,
+      },
+    },
+  ];
+
+  if (reason) {
+    blocks.push({
+      type: "section",
+      text: { type: "mrkdwn", text: `*Reason:* ${reason}` },
+    });
+  }
+
+  blocks.push({
+    type: "section",
+    text: { type: "mrkdwn", text: cdnLine },
+  });
+
+  await getLogClient(client).chat.postMessage({
+    channel: LOG_CHANNEL,
+    text: `Messages purged in <#${channel}>`,
+    blocks,
+  });
+}
diff --git a/lib/public-logger.js b/lib/public-logger.js
index 1ad652b..4193027 100644
--- a/lib/public-logger.js
+++ b/lib/public-logger.js
@@ -50,3 +50,21 @@ export async function publicLogThread(client, { channel, messages, deletedBy, ap
     ],
   });
 }
+
+export async function publicLogPurge(client, { channel, count, deletedBy, api = false }) {
+  if (!c) return;
+
+  await getLogClient(client).chat.postMessage({
+    channel: c,
+    text: `Messages purged in <#${channel}>${api ? " (via API)" : ""}`,
+    blocks: [
+      {
+        type: "section",
+        text: {
+          type: "mrkdwn",
+          text: `:broom: <@${deletedBy}> purged ${count} message${count === 1 ? "" : "s"} in <#${channel}>.${apiTag(api)}`,
+        },
+      },
+    ],
+  });
+}
diff --git a/tests/purge-command.test.js b/tests/purge-command.test.js
new file mode 100644
index 0000000..9324d53
--- /dev/null
+++ b/tests/purge-command.test.js
@@ -0,0 +1,148 @@
+import { beforeEach, expect, mock, test } from "bun:test";
+
+const cascade = mock(async () => {});
+const deleteBatch = mock(async (_client, _logger, _channel, messages) => ({
+  dc: messages.length,
+  ec: 0,
+}));
+const noop = async () => {};
+
+mock.module("../lib/perms.js", () => ({ canManage: async () => true }));
+mock.module("../lib/ratelimiter.js", () => ({
+  RateLimiter: class {
+    deleteBatch = deleteBatch;
+  },
+}));
+mock.module("../lib/logger.js", () => ({ logPurge: noop, notifyPurgeDeletion: noop }));
+mock.module("../lib/public-logger.js", () => ({ publicLogPurge: noop }));
+mock.module("../lib/purge.js", () => ({ purge: cascade }));
+
+const { views } = await import("../lib/commands/purge.js");
+const handleView = views[0].handleView;
+const parent = { ts: "1690000000.000001", reply_count: 2 };
+const reply = { ts: "1690000000.000002", thread_ts: parent.ts };
+const older = { ts: "1689999999.000001" };
+
+beforeEach(() => {
+  cascade.mockClear();
+  deleteBatch.mockClear();
+});
+
+async function submit({
+  includeThreads = true,
+  exclude = "",
+  range = "",
+  countRangeInTotal = false,
+  historyPages = [{ messages: [parent, older] }],
+  repliesPages = [{ messages: [parent, reply] }],
+} = {}) {
+  let historyPage = 0;
+  let repliesPage = 0;
+  const userClient = {
+    conversations: {
+      history: mock(async () => historyPages[historyPage++]),
+      replies: mock(async () => repliesPages[repliesPage++]),
+    },
+  };
+  const client = {
+    conversations: {
+      history: mock(async () => {
+        throw new Error("Bot is not a channel member");
+      }),
+    },
+    chat: { postEphemeral: mock(noop) },
+  };
+  await handleView({
+    view: {
+      private_metadata: JSON.stringify({ channel: "C123", count: 1, includeThreads }),
+      state: {
+        values: {
+          reason: { reason_input: { value: "Test purge" } },
+          exclude: { exclude_input: { value: exclude } },
+          range: { range_input: { value: range } },
+          range_count: {
+            range_count_checkbox: {
+              selected_options: countRangeInTotal ? [{ value: "count" }] : [],
+            },
+          },
+        },
+      },
+    },
+    body: { user: { id: "U123" } },
+    client,
+    context: { userClient },
+    logger: { info() {}, warn() {}, error() {} },
+  });
+  return { client, userClient };
+}
+
+function expectOlderDeleted() {
+  expect(cascade).not.toHaveBeenCalled();
+  expect(deleteBatch).toHaveBeenCalledTimes(1);
+  expect(deleteBatch.mock.calls[0][3]).toEqual([older]);
+}
+
+test("excluded reply protects its whole thread and does not consume the count", async () => {
+  await submit({ exclude: "https://slack.com/archives/C123/p1690000000000002" });
+  expectOlderDeleted();
+});
+
+test("kept range protects replies even when the parent is outside the range", async () => {
+  await submit({ range: "1690000000.000002\n1690000000.000003" });
+  expectOlderDeleted();
+});
+
+test("reply protection checks every replies page and continues across history pages", async () => {
+  const { userClient } = await submit({
+    exclude: reply.ts,
+    historyPages: [
+      { messages: [parent], response_metadata: { next_cursor: "older-history" } },
+      { messages: [older] },
+    ],
+    repliesPages: [
+      { messages: [parent], response_metadata: { next_cursor: "more-replies" } },
+      { messages: [reply] },
+    ],
+  });
+  expectOlderDeleted();
+  expect(userClient.conversations.replies.mock.calls[1][0].cursor).toBe("more-replies");
+  expect(userClient.conversations.history.mock.calls[1][0].cursor).toBe("older-history");
+});
+
+test("unprotected threads still cascade using the user client", async () => {
+  const { userClient } = await submit({ exclude: "1690000001.000001" });
+  expect(cascade).toHaveBeenCalledTimes(1);
+  expect(cascade.mock.calls[0][0]).toBe(userClient);
+  expect(cascade.mock.calls[0][3]).toBe(parent.ts);
+  expect(deleteBatch).not.toHaveBeenCalled();
+});
+
+test("no protections avoids the extra replies scan and history uses the user token", async () => {
+  const { client, userClient } = await submit();
+  expect(userClient.conversations.history).toHaveBeenCalledTimes(1);
+  expect(client.conversations.history).not.toHaveBeenCalled();
+  expect(userClient.conversations.replies).not.toHaveBeenCalled();
+  expect(cascade).toHaveBeenCalledTimes(1);
+});
+
+test("non-cascading purges do not skip parents of protected replies", async () => {
+  const { userClient } = await submit({ includeThreads: false, exclude: reply.ts });
+  expect(userClient.conversations.replies).not.toHaveBeenCalled();
+  expect(cascade).not.toHaveBeenCalled();
+  expect(deleteBatch.mock.calls[0][3]).toEqual([parent]);
+});
+
+test("counting a protected parent range still consumes the fixed window", async () => {
+  await submit({
+    range: `${parent.ts}\n${reply.ts}`,
+    countRangeInTotal: true,
+  });
+  expect(cascade).not.toHaveBeenCalled();
+  expect(deleteBatch).not.toHaveBeenCalled();
+});
+
+test("a failed replies scan aborts before deleting any messages", async () => {
+  await expect(submit({ exclude: reply.ts, repliesPages: [] })).rejects.toThrow();
+  expect(cascade).not.toHaveBeenCalled();
+  expect(deleteBatch).not.toHaveBeenCalled();
+});