diff --git a/inc/Abilities/Fetch/FetchEmailAbility.php b/inc/Abilities/Fetch/FetchEmailAbility.php
index e17a762f35..be66961715 100644
--- a/inc/Abilities/Fetch/FetchEmailAbility.php
+++ b/inc/Abilities/Fetch/FetchEmailAbility.php
@@ -20,10 +20,16 @@
 
 use DataMachine\Abilities\PermissionHelper;
 use DataMachine\Core\Steps\Fetch\Handlers\Email\EmailAuth;
+use DataMachine\Abilities\Fetch\Traits\HasApplyKeywordSearch;
+use DataMachine\Abilities\Fetch\Traits\HasApplyKeywordExclusion;
 
 defined( 'ABSPATH' ) || exit;
 
 class FetchEmailAbility {
+	use HasApplyKeywordSearch;
+	use HasApplyKeywordExclusion;
+
+	private const KEYWORD_SCAN_LIMIT = 500;
 
 	private static bool $registered = false;
 
@@ -59,9 +65,15 @@ private function registerAbilities(): void {
 								'description' => __( 'Mail folder to fetch from', 'data-machine' ),
 							),
 							'search_criteria'      => array(
-								'type'        => 'string',
+								'anyOf'       => array(
+									array( 'type' => 'string' ),
+									array(
+										'type'  => 'array',
+										'items' => array( 'type' => 'string' ),
+									),
+								),
 								'default'     => 'UNSEEN',
-								'description' => __( 'IMAP search string (UNSEEN, ALL, FROM "x", SINCE "1-Mar-2026")', 'data-machine' ),
+								'description' => __( 'IMAP search string (UNSEEN, ALL, FROM "x", SINCE "1-Mar-2026"), or a list of strings whose results are merged (OR)', 'data-machine' ),
 							),
 							'max_messages'         => array(
 								'type'        => 'integer',
@@ -73,6 +85,16 @@ private function registerAbilities(): void {
 								'default'     => 0,
 								'description' => __( 'Number of messages to skip (for pagination)', 'data-machine' ),
 							),
+							'search'               => array(
+								'type'        => 'string',
+								'default'     => '',
+								'description' => __( 'Comma-separated keywords; a message matches if its subject or body contains any of them', 'data-machine' ),
+							),
+							'exclude_keywords'     => array(
+								'type'        => 'string',
+								'default'     => '',
+								'description' => __( 'Comma-separated keywords; a message is skipped if its subject or body contains any of them', 'data-machine' ),
+							),
 							'uid'                  => array(
 								'type'        => 'integer',
 								'default'     => 0,
@@ -215,10 +237,11 @@ public function executeWithContext( array $input, array $context ): array|\WP_Er
 			);
 		}
 
-		// Search for messages.
-		$message_ids = imap_search( $connection, $config['search_criteria'], SE_UID );
+		// Search for messages. c-client has no OR support, so callers pass
+		// one criteria string per alternative and the UIDs are merged here.
+		$message_ids = $this->searchUids( $connection, $config['search_criteria'] );
 
-		if ( false === $message_ids || empty( $message_ids ) ) {
+		if ( empty( $message_ids ) ) {
 			imap_close( $connection );
 			return array(
 				'success' => true,
@@ -233,36 +256,21 @@ public function executeWithContext( array $input, array $context ): array|\WP_Er
 			);
 		}
 
-		// Most recent first.
-		$message_ids   = array_reverse( $message_ids );
-		$total_matches = count( $message_ids );
-		$offset        = (int) $config['offset'];
-		$max           = (int) $config['max_messages'];
+		$offset    = max( 0, (int) $config['offset'] );
+		$max       = (int) $config['max_messages'];
+		$selection = $this->selectUids( $connection, $message_ids, $config, $offset, $max, $logs );
 
-		// Apply pagination.
-		if ( $offset > 0 ) {
-			$message_ids = array_slice( $message_ids, $offset );
-		}
-		$message_ids = array_slice( $message_ids, 0, $max );
-		$has_more    = ( $offset + count( $message_ids ) ) < $total_matches;
+		$message_ids   = $selection['uids'];
+		$total_matches = $selection['total_matches'];
+		$has_more      = $selection['has_more'];
 
 		$items = array();
-
-		if ( $config['headers_only'] ) {
-			// Fast header-only mode — no body parsing, no structure fetching.
-			foreach ( $message_ids as $uid ) {
-				$item = $this->fetchHeaders( $connection, $uid );
-				if ( null !== $item ) {
-					$items[] = $item;
-				}
-			}
-		} else {
-			// Full mode — bodies + attachments.
-			foreach ( $message_ids as $uid ) {
-				$item = $this->fetchMessage( $connection, $uid, $config );
-				if ( null !== $item ) {
-					$items[] = $item;
-				}
+		foreach ( $message_ids as $uid ) {
+			$item = $config['headers_only']
+				? $this->fetchHeaders( $connection, (int) $uid )
+				: $this->fetchMessage( $connection, (int) $uid, $config );
+			if ( null !== $item ) {
+				$items[] = $item;
 			}
 		}
 
@@ -302,13 +310,129 @@ public function executeWithContext( array $input, array $context ): array|\WP_Er
 		);
 	}
 
+	/**
+	 * Run one IMAP search per criteria string and merge the UIDs newest first.
+	 *
+	 * @param \IMAP\Connection        $connection IMAP connection.
+	 * @param string|string[] $criteria   One criteria string or a list of OR'd alternatives.
+	 * @return int[] Unique UIDs, newest first.
+	 */
+	private function searchUids( $connection, string|array $criteria ): array {
+		$uids = array();
+		foreach ( (array) $criteria as $single ) {
+			$found = imap_search( $connection, (string) $single, SE_UID );
+			if ( is_array( $found ) ) {
+				$uids = array_merge( $uids, $found );
+			}
+		}
+
+		$uids = array_values( array_unique( array_map( 'intval', $uids ) ) );
+		// UIDs increase with arrival order within a folder.
+		rsort( $uids, SORT_NUMERIC );
+
+		return $uids;
+	}
+
+	/**
+	 * Pick the page of UIDs to fetch, applying keyword filters before the cap.
+	 *
+	 * Without keywords this is plain pagination and fetches nothing extra.
+	 * With keywords it walks UIDs newest first, decides on the subject when it
+	 * can, reads the body text (peek, no attachments) only when it must, and
+	 * stops once the requested page is filled or the scan limit is reached.
+	 *
+	 * @param \IMAP\Connection $connection IMAP connection.
+	 * @param int[]    $uids       Candidate UIDs, newest first.
+	 * @param array    $config     Normalized config.
+	 * @param int      $offset     Page offset.
+	 * @param int      $max        Page size.
+	 * @param array    $logs       Log entries, appended by reference.
+	 * @return array{uids: int[], total_matches: int, has_more: bool}
+	 */
+	private function selectUids( $connection, array $uids, array $config, int $offset, int $max, array &$logs ): array {
+		$include = trim( (string) ( $config['search'] ?? '' ) );
+		$exclude = trim( (string) ( $config['exclude_keywords'] ?? '' ) );
+
+		if ( '' === $include && '' === $exclude ) {
+			$total = count( $uids );
+			return array(
+				'uids'          => array_slice( $uids, $offset, max( 0, $max ) ),
+				'total_matches' => $total,
+				'has_more'      => ( $offset + $max ) < $total,
+			);
+		}
+
+		$needed  = $offset + max( 1, $max );
+		$limit   = (int) apply_filters( 'datamachine_email_keyword_scan_limit', self::KEYWORD_SCAN_LIMIT );
+		$scanned = 0;
+		$matched = array();
+
+		foreach ( $uids as $uid ) {
+			if ( $scanned >= $limit || count( $matched ) >= $needed ) {
+				break;
+			}
+			++$scanned;
+
+			if ( $this->messageMatchesKeywords( $connection, $uid, $include, $exclude ) ) {
+				$matched[] = $uid;
+			}
+		}
+
+		$unscanned = $scanned < count( $uids );
+		if ( $unscanned && $scanned >= $limit && count( $matched ) < $needed ) {
+			$logs[] = array(
+				'level'   => 'warning',
+				'message' => sprintf( 'Email Fetch: keyword scan stopped at the %d-message limit; narrow the search criteria or timeframe to reach older matches', $limit ),
+			);
+		}
+
+		return array(
+			'uids'          => array_slice( $matched, $offset, max( 0, $max ) ),
+			// Lower bound when the scan stopped early.
+			'total_matches' => count( $matched ),
+			'has_more'      => $unscanned || count( $matched ) > $offset + $max,
+		);
+	}
+
+	/**
+	 * Decide whether one message passes the include/exclude keyword filters.
+	 *
+	 * Subject first; the body text is only read when the subject cannot decide.
+	 *
+	 * @param \IMAP\Connection $connection IMAP connection.
+	 * @param int      $uid        Message UID.
+	 * @param string   $include_keywords Comma-separated include keywords (OR).
+	 * @param string   $exclude    Comma-separated exclude keywords.
+	 * @return bool
+	 */
+	private function messageMatchesKeywords( $connection, int $uid, string $include_keywords, string $exclude ): bool {
+		$headers = $this->fetchHeaders( $connection, $uid );
+		if ( null === $headers ) {
+			return false;
+		}
+
+		$subject = (string) ( $headers['title'] ?? '' );
+		if ( $this->applyKeywordExclusion( $subject, $exclude ) ) {
+			return false;
+		}
+
+		// Subject already includes a keyword and there is no body exclusion to check.
+		if ( '' === $exclude && '' !== $include_keywords && $this->applyKeywordSearch( $subject, $include_keywords ) ) {
+			return true;
+		}
+
+		$text = $subject . ' ' . $this->fetchBody( $connection, $uid, true );
+
+		return $this->applyKeywordSearch( $text, $include_keywords ) && ! $this->applyKeywordExclusion( $text, $exclude );
+	}
+
 	/**
 	 * Fetch headers only for a message (fast list mode).
 	 *
 	 * No body parsing, no structure fetching, no attachment scanning.
 	 * Just From, Subject, Date, UID, and basic flags.
 	 *
-	 * @param resource $connection IMAP connection.
+	 * @param \IMAP\Connection $connection IMAP connection.
 	 * @param int      $uid        Message UID.
 	 * @return array|null Header data or null on failure.
 	 */
@@ -368,7 +492,7 @@ private function fetchHeaders( $connection, int $uid ): ?array {
 	/**
 	 * Fetch a single message with full body by UID.
 	 *
-	 * @param resource $connection IMAP connection.
+	 * @param \IMAP\Connection $connection IMAP connection.
 	 * @param int      $uid        Message UID.
 	 * @param array    $config     Fetch configuration.
 	 * @return array|null Message data or null on failure.
@@ -481,7 +605,7 @@ private function fetchMessage( $connection, int $uid, array $config ): ?array {
 	/**
 	 * Fetch message body, preferring plain text.
 	 *
-	 * @param resource $connection IMAP connection.
+	 * @param \IMAP\Connection $connection IMAP connection.
 	 * @param int      $uid        Message UID.
 	 * @return string Message body text.
 	 */
@@ -495,7 +619,7 @@ private function fetchBody( $connection, int $uid, bool $peek ): string {
 		// Simple single-part message.
 		if ( empty( $structure->parts ) ) {
 			$body = imap_fetchbody( $connection, $uid, '1', $fetch_flags );
-			return $this->decodeBody( $body, $structure->encoding ?? 0 );
+			return $this->decodeBody( (string) $body, $structure->encoding ?? 0 );
 		}
 
 		// Multipart — find text/plain first, then text/html.
@@ -508,7 +632,7 @@ private function fetchBody( $connection, int $uid, bool $peek ): string {
 			if ( 0 === ( $part->type ?? -1 ) ) {
 				$subtype = strtolower( $part->subtype ?? '' );
 				$body    = imap_fetchbody( $connection, $uid, $part_number, $fetch_flags );
-				$decoded = $this->decodeBody( $body, $part->encoding ?? 0 );
+				$decoded = $this->decodeBody( (string) $body, $part->encoding ?? 0 );
 
 				if ( 'plain' === $subtype && empty( $plain_body ) ) {
 					$plain_body = $decoded;
@@ -525,7 +649,7 @@ private function fetchBody( $connection, int $uid, bool $peek ): string {
 					if ( 0 === ( $sub_part->type ?? -1 ) ) {
 						$subtype = strtolower( $sub_part->subtype ?? '' );
 						$body    = imap_fetchbody( $connection, $uid, $sub_number, $fetch_flags );
-						$decoded = $this->decodeBody( $body, $sub_part->encoding ?? 0 );
+						$decoded = $this->decodeBody( (string) $body, $sub_part->encoding ?? 0 );
 
 						if ( 'plain' === $subtype && empty( $plain_body ) ) {
 							$plain_body = $decoded;
@@ -628,7 +752,7 @@ private function downloadAttachment( $connection, int $uid, array $attachment_in
 			return null;
 		}
 
-		$decoded = $this->decodeBody( $body, $attachment_info['encoding'] );
+		$decoded = $this->decodeBody( (string) $body, $attachment_info['encoding'] );
 		if ( empty( $decoded ) ) {
 			return null;
 		}
@@ -706,6 +830,8 @@ private function normalizeConfig( array $input ): array {
 			'headers_only'         => false,
 			'mark_as_read'         => false,
 			'download_attachments' => false,
+			'search'               => '',
+			'exclude_keywords'     => '',
 		);
 
 		return array_merge( $defaults, $input );
diff --git a/inc/Core/Steps/Fetch/Handlers/Email/Email.php b/inc/Core/Steps/Fetch/Handlers/Email/Email.php
index 4f39363c67..83eb296d8b 100644
--- a/inc/Core/Steps/Fetch/Handlers/Email/Email.php
+++ b/inc/Core/Steps/Fetch/Handlers/Email/Email.php
@@ -61,6 +61,8 @@ protected function executeFetch( array $config, ExecutionContext $context ): arr
 			'folder'               => $config['folder'] ?? 'INBOX',
 			'search_criteria'      => $search_criteria,
 			'max_messages'         => (int) ( $config['max_messages'] ?? 10 ),
+			'search'               => (string) ( $config['search'] ?? '' ),
+			'exclude_keywords'     => (string) ( $config['exclude_keywords'] ?? '' ),
 			'mark_as_read'         => ! empty( $config['mark_as_read'] ),
 			'download_attachments' => ! empty( $config['download_attachments'] ),
 		);
@@ -111,34 +113,62 @@ protected function executeFetch( array $config, ExecutionContext $context ): arr
 	/**
 	 * Build IMAP search criteria from handler config.
 	 *
-	 * Starts with the base search criteria and augments with
-	 * from_filter and timeframe_limit settings.
+	 * Starts with the base search criteria and augments it with the sender,
+	 * subject and timeframe filters. PHP's c-client imap_search() rejects the
+	 * IMAP OR keyword, so comma-separated sender/subject values produce one
+	 * criteria string per combination; the ability merges and dedupes the UIDs.
 	 *
 	 * @param array $config Handler configuration.
-	 * @return string IMAP search criteria string.
+	 * @return string|string[] One criteria string, or a list of OR'd alternatives.
 	 */
-	private function buildSearchCriteria( array $config ): string {
-		$criteria = $config['search_criteria'] ?? 'UNSEEN';
+	private function buildSearchCriteria( array $config ): string|array {
+		$base     = (string) ( $config['search_criteria'] ?? 'UNSEEN' );
+		$senders  = $this->splitFilterValues( (string) ( $config['from_filter'] ?? '' ) );
+		$subjects = $this->splitFilterValues( (string) ( $config['subject_filter'] ?? '' ) );
 
-		// Augment with sender filter.
-		if ( ! empty( $config['from_filter'] ) ) {
-			$criteria .= ' FROM "' . $config['from_filter'] . '"';
-		}
-
-		// Augment with subject filter.
-		if ( ! empty( $config['subject_filter'] ) ) {
-			$criteria .= ' SUBJECT "' . $config['subject_filter'] . '"';
+		$since = '';
+		if ( ! empty( $config['timeframe_limit'] ) && 'all_time' !== $config['timeframe_limit'] ) {
+			$since_date = $this->timeframeToDate( (string) $config['timeframe_limit'] );
+			if ( null !== $since_date ) {
+				$since = ' SINCE "' . $since_date . '"';
+			}
 		}
 
-		// Augment with timeframe filter.
-		if ( ! empty( $config['timeframe_limit'] ) && 'all_time' !== $config['timeframe_limit'] ) {
-			$since_date = $this->timeframeToDate( $config['timeframe_limit'] );
-			if ( $since_date ) {
-				$criteria .= ' SINCE "' . $since_date . '"';
+		$searches = array();
+		foreach ( ( empty( $senders ) ? array( '' ) : $senders ) as $sender ) {
+			foreach ( ( empty( $subjects ) ? array( '' ) : $subjects ) as $subject ) {
+				$search = $base;
+				if ( '' !== $sender ) {
+					$search .= ' FROM "' . $this->quoteImapString( $sender ) . '"';
+				}
+				if ( '' !== $subject ) {
+					$search .= ' SUBJECT "' . $this->quoteImapString( $subject ) . '"';
+				}
+				$searches[] = $search . $since;
 			}
 		}
 
-		return $criteria;
+		return 1 === count( $searches ) ? $searches[0] : $searches;
+	}
+
+	/**
+	 * Split a comma-separated filter value into trimmed, non-empty parts.
+	 *
+	 * @param string $value Raw filter value.
+	 * @return string[]
+	 */
+	private function splitFilterValues( string $value ): array {
+		return array_values( array_filter( array_map( 'trim', explode( ',', $value ) ), static fn( string $value ): bool => '' !== $value ) );
+	}
+
+	/**
+	 * Escape a value for use inside an IMAP quoted string.
+	 *
+	 * @param string $value Raw value.
+	 * @return string
+	 */
+	private function quoteImapString( string $value ): string {
+		return str_replace( array( '\\', '"' ), array( '\\\\', '\\"' ), $value );
 	}
 
 	/**
@@ -148,20 +178,11 @@ private function buildSearchCriteria( array $config ): string {
 	 * @return string|null IMAP-formatted date or null.
 	 */
 	private function timeframeToDate( string $timeframe ): ?string {
-		$days_map = array(
-			'24_hours' => 1,
-			'7_days'   => 7,
-			'30_days'  => 30,
-			'90_days'  => 90,
-			'6_months' => 180,
-			'1_year'   => 365,
-		);
-
-		$days = $days_map[ $timeframe ] ?? null;
-		if ( null === $days ) {
+		$timestamp = apply_filters( 'datamachine_timeframe_limit', null, $timeframe );
+		if ( ! is_numeric( $timestamp ) ) {
+			do_action( 'datamachine_log', 'warning', 'Unknown email timeframe_limit: ' . $timeframe );
 			return null;
 		}
-
-		return gmdate( 'j-M-Y', strtotime( "-{$days} days" ) );
+		return gmdate( 'j-M-Y', (int) $timestamp );
 	}
 }
diff --git a/inc/Core/Steps/Fetch/Handlers/Email/EmailFetchSettings.php b/inc/Core/Steps/Fetch/Handlers/Email/EmailFetchSettings.php
index 91496de35c..369da821ed 100644
--- a/inc/Core/Steps/Fetch/Handlers/Email/EmailFetchSettings.php
+++ b/inc/Core/Steps/Fetch/Handlers/Email/EmailFetchSettings.php
@@ -44,12 +44,12 @@ public static function get_fields(): array {
 			'from_filter'          => array(
 				'type'        => 'text',
 				'label'       => __( 'From Filter', 'data-machine' ),
-				'description' => __( 'Only fetch emails from this sender address.', 'data-machine' ),
+				'description' => __( 'Only fetch emails from these sender addresses; separate multiple values with commas (OR).', 'data-machine' ),
 			),
 			'subject_filter'       => array(
 				'type'        => 'text',
 				'label'       => __( 'Subject Filter', 'data-machine' ),
-				'description' => __( 'Only fetch emails with this text in the subject.', 'data-machine' ),
+				'description' => __( 'Only fetch emails with any of these subject texts; separate multiple values with commas (OR).', 'data-machine' ),
 			),
 			'mark_as_read'         => array(
 				'type'        => 'checkbox',
diff --git a/tests/email-fetch-filters-smoke.php b/tests/email-fetch-filters-smoke.php
new file mode 100644
index 0000000000..cc748812ba
--- /dev/null
+++ b/tests/email-fetch-filters-smoke.php
@@ -0,0 +1,277 @@
+<?php
+/**
+ * Email fetch filters: shared keyword filters, timeframe mapping, and OR'd
+ * sender/subject searches, against a deterministic in-memory IMAP mailbox.
+ *
+ * @package DataMachine\Tests
+ */
+
+namespace {
+	define( 'ABSPATH', __DIR__ . '/' );
+	foreach ( array(
+		'FT_UID'  => 1,
+		'FT_PEEK' => 2,
+		'SE_UID'  => 1,
+	) as $constant => $value ) {
+		if ( ! defined( $constant ) ) {
+			define( $constant, $value );
+		}
+	}
+
+	$GLOBALS['filters']  = array();
+	$GLOBALS['logged']   = array();
+	$GLOBALS['imap_log'] = array();
+
+	function apply_filters( string $hook, mixed $value, mixed ...$args ): mixed {
+		foreach ( $GLOBALS['filters'][ $hook ] ?? array() as $callback ) {
+			$value = $callback( $value, ...$args );
+		}
+		return $value;
+	}
+	function add_filter( string $hook, callable $callback ): void {
+		$GLOBALS['filters'][ $hook ][] = $callback;
+	}
+	function do_action( string $hook, mixed ...$args ): void {
+		$GLOBALS['logged'][] = array( $hook, $args );
+	}
+	function __( string $text, string $domain = '' ): string {
+		return $text;
+	}
+
+	// Same contract as the shared filter in inc/Engine/Filters/Handlers.php.
+	add_filter(
+		'datamachine_timeframe_limit',
+		static function ( $default_value, $timeframe ) {
+			$map = array(
+				'24_hours' => '-24 hours',
+				'72_hours' => '-72 hours',
+				'7_days'   => '-7 days',
+				'30_days'  => '-30 days',
+			);
+			return isset( $map[ $timeframe ] ) ? strtotime( $map[ $timeframe ] ) : null;
+		}
+	);
+
+	/*
+	 * Fixture mailbox, keyed by UID. Bodies are only reachable through
+	 * imap_fetchbody so the test can count how many bodies were read.
+	 */
+	$GLOBALS['mailbox'] = array(
+		101 => array( 'Album out today', 'New single from a band', 'pr@label.test' ),
+		102 => array( 'High Water Fest Announces 2027 Lineup', 'Jack Johnson and Mt. Joy headline', 'sue@thesyn.test' ),
+		103 => array( 'WordCamp US Lineup Keeps Growing', 'Speakers announced', 'noreply@wordpress.test' ),
+		104 => array( 'Press release', 'Rooster Walk festival lineup drops Oct 20', 'info@roosterwalk.test' ),
+		105 => array( 'Invoice 55', 'Pay by Friday', 'billing@vendor.test' ),
+		106 => array( 'Big Ears Festival Expands 2027 Lineup', 'The Roots join the bill', 'chad@pitchperfect.test' ),
+	);
+}
+
+namespace DataMachine\Abilities\Fetch {
+	function imap_search( $connection, string $criteria, int $flags ): array|false {
+		$GLOBALS['imap_log'][] = array( 'search', $criteria );
+		return $GLOBALS['search_results'][ $criteria ] ?? array_keys( $GLOBALS['mailbox'] );
+	}
+	function imap_msgno( $connection, int $uid ): int {
+		return isset( $GLOBALS['mailbox'][ $uid ] ) ? $uid : 0;
+	}
+	function imap_headerinfo( $connection, int $msgno ): object {
+		$GLOBALS['imap_log'][]    = array( 'headers', $msgno );
+		list( $subject, , $from ) = $GLOBALS['mailbox'][ $msgno ];
+		list( $mailbox, $host )   = explode( '@', $from );
+		return (object) array(
+			'subject'    => $subject,
+			'from'       => array(
+				(object) array(
+					'mailbox' => $mailbox,
+					'host'    => $host,
+				),
+			),
+			'date'       => '2026-10-13 10:00:00',
+			'message_id' => '<' . $msgno . '@fixture.test>',
+		);
+	}
+	function imap_utf8( string $value ): string {
+		return $value;
+	}
+	function imap_fetch_overview( $connection, string $uid, int $flags ): array {
+		return array( (object) array( 'size' => 100 ) );
+	}
+	function imap_fetchstructure( $connection, int $uid, int $flags ): object {
+		return (object) array( 'encoding' => 0 );
+	}
+	function imap_fetchbody( $connection, int $uid, string $section, int $flags ): string {
+		$GLOBALS['imap_log'][] = array( 'body', $uid, $flags );
+		return $GLOBALS['mailbox'][ $uid ][1];
+	}
+}
+
+namespace DataMachine\Core\Steps {
+	trait HandlerRegistrationTrait {
+	}
+}
+
+namespace DataMachine\Core\Steps\Fetch\Handlers {
+	class FetchHandler {
+	}
+}
+
+namespace {
+	$root = dirname( __DIR__ );
+	require_once $root . '/inc/Abilities/Fetch/Traits/HasApplyKeywordSearch.php';
+	require_once $root . '/inc/Abilities/Fetch/Traits/HasApplyKeywordExclusion.php';
+	require_once $root . '/inc/Abilities/Fetch/FetchEmailAbility.php';
+	require_once $root . '/inc/Core/Steps/Fetch/Handlers/Email/Email.php';
+
+	$failures = array();
+	$assert   = static function ( bool $condition, string $message ) use ( &$failures ): void {
+		echo ( $condition ? 'PASS: ' : 'FAIL: ' ) . $message . "\n";
+		if ( ! $condition ) {
+			$failures[] = $message;
+		}
+	};
+
+	$ability   = ( new ReflectionClass( DataMachine\Abilities\Fetch\FetchEmailAbility::class ) )->newInstanceWithoutConstructor();
+	$handler   = ( new ReflectionClass( DataMachine\Core\Steps\Fetch\Handlers\Email\Email::class ) )->newInstanceWithoutConstructor();
+	$call      = static function ( object $target, string $method, mixed ...$args ): mixed {
+		$reflection = new ReflectionMethod( $target, $method );
+		$reflection->setAccessible( true );
+		return $reflection->invokeArgs( $target, $args );
+	};
+	$select    = static function ( array $config, int $offset, int $max ) use ( $ability ): array {
+		$GLOBALS['imap_log'] = array();
+		$logs                = array();
+		$uids                = array( 106, 105, 104, 103, 102, 101 );
+		$reflection          = new ReflectionMethod( $ability, 'selectUids' );
+		$reflection->setAccessible( true );
+		$result         = $reflection->invokeArgs( $ability, array( null, $uids, $config, $offset, $max, &$logs ) );
+		$result['logs'] = $logs;
+		return $result;
+	};
+	$count_ops = static function ( string $op ): int {
+		return count( array_filter( $GLOBALS['imap_log'], static fn( $entry ) => $op === $entry[0] ) );
+	};
+
+	// No keywords: plain pagination, nothing fetched during selection.
+	$plain = $select( array(
+		'search'           => '',
+		'exclude_keywords' => '',
+	), 0, 2 );
+	$assert( array( 106, 105 ) === $plain['uids'], 'without keywords the newest page is selected' );
+	$assert( 6 === $plain['total_matches'] && true === $plain['has_more'], 'without keywords totals and has_more are exact' );
+	$assert( 0 === $count_ops( 'headers' ) && 0 === $count_ops( 'body' ), 'without keywords selection fetches no headers or bodies' );
+
+	// Include keywords: subject or body matches, filtered before the cap.
+	$lineup = $select( array(
+		'search'           => 'lineup',
+		'exclude_keywords' => '',
+	), 0, 10 );
+	$assert( array( 106, 104, 103, 102 ) === $lineup['uids'], 'include keywords match subject or body text' );
+	$assert( 3 === $count_ops( 'body' ), 'bodies are only read when the subject does not match' );
+	foreach ( $GLOBALS['imap_log'] as $entry ) {
+		if ( 'body' === $entry[0] && 0 === ( $entry[2] & FT_PEEK ) ) {
+			$assert( false, 'keyword body reads must use FT_PEEK' );
+		}
+	}
+
+	// Exclude keywords on top of include keywords.
+	$festivals = $select( array(
+		'search'           => 'lineup, festival',
+		'exclude_keywords' => 'WordCamp',
+	), 0, 10 );
+	$assert( array( 106, 104, 102 ) === $festivals['uids'], 'exclude keywords drop matching messages' );
+
+	$no_invoices = $select( array(
+		'search'           => '',
+		'exclude_keywords' => 'Friday',
+	), 0, 10 );
+	$assert( ! in_array( 105, $no_invoices['uids'], true ), 'exclude keywords also check body text' );
+
+	// Matches are not starved by the cap: one match per page, newest first.
+	$page_one = $select( array(
+		'search'           => 'lineup',
+		'exclude_keywords' => '',
+	), 0, 1 );
+	$assert( array( 106 ) === $page_one['uids'] && true === $page_one['has_more'], 'page one returns the newest match' );
+	$assert( 1 === $count_ops( 'headers' ), 'scan stops as soon as the page is filled' );
+	$page_two = $select( array(
+		'search'           => 'lineup',
+		'exclude_keywords' => '',
+	), 1, 1 );
+	$assert( array( 104 ) === $page_two['uids'], 'offset skips earlier matches, not earlier messages' );
+
+	// Scan limit is bounded and logged.
+	add_filter( 'datamachine_email_keyword_scan_limit', static fn() => 2 );
+	$limited = $select( array(
+		'search'           => 'Rooster',
+		'exclude_keywords' => '',
+	), 0, 5 );
+	$assert( array() === $limited['uids'] && true === $limited['has_more'], 'scan limit stops early and reports has_more' );
+	$assert( 2 === $count_ops( 'headers' ), 'scan limit caps candidates read' );
+	$assert( 1 === count( $limited['logs'] ) && 'warning' === $limited['logs'][0]['level'], 'hitting the scan limit logs a warning' );
+	unset( $GLOBALS['filters']['datamachine_email_keyword_scan_limit'] );
+
+	// Multiple criteria: merged, deduped, newest first.
+	$GLOBALS['search_results'] = array(
+		'A' => array( 101, 104 ),
+		'B' => array( 104, 106, 102 ),
+		'C' => false,
+	);
+	$merged                    = $call( $ability, 'searchUids', null, array( 'A', 'B', 'C' ) );
+	$assert( array( 106, 104, 102, 101 ) === $merged, 'OR searches merge, dedupe and sort newest first' );
+	$single = $call( $ability, 'searchUids', null, 'A' );
+	$assert( array( 104, 101 ) === $single, 'a single criteria string still works' );
+
+	// Handler criteria building.
+	$criteria = $call( $handler, 'buildSearchCriteria', array(
+		'search_criteria' => 'ALL',
+		'timeframe_limit' => 'all_time',
+	) );
+	$assert( 'ALL' === $criteria, 'all_time adds no SINCE' );
+
+	$criteria = $call( $handler, 'buildSearchCriteria', array(
+		'search_criteria' => 'ALL',
+		'timeframe_limit' => '72_hours',
+	) );
+	$assert( 'ALL SINCE "' . gmdate( 'j-M-Y', strtotime( '-72 hours' ) ) . '"' === $criteria, '72_hours maps to a SINCE date through the shared filter' );
+
+	$GLOBALS['logged'] = array();
+	$criteria          = $call( $handler, 'buildSearchCriteria', array(
+		'search_criteria' => 'ALL',
+		'timeframe_limit' => '90_days',
+	) );
+	$assert( 'ALL' === $criteria, 'unknown timeframe adds no SINCE' );
+	$assert( 1 === count( $GLOBALS['logged'] ) && 'warning' === $GLOBALS['logged'][0][1][0], 'unknown timeframe logs a warning' );
+
+	$criteria = $call(
+		$handler,
+		'buildSearchCriteria',
+		array(
+			'search_criteria' => 'ALL',
+			'from_filter'     => 'sue@thesyn.test, chad@pitchperfect.test',
+			'subject_filter'  => 'lineup,festival',
+			'timeframe_limit' => '7_days',
+		)
+	);
+	$since    = ' SINCE "' . gmdate( 'j-M-Y', strtotime( '-7 days' ) ) . '"';
+	$assert(
+		array(
+			'ALL FROM "sue@thesyn.test" SUBJECT "lineup"' . $since,
+			'ALL FROM "sue@thesyn.test" SUBJECT "festival"' . $since,
+			'ALL FROM "chad@pitchperfect.test" SUBJECT "lineup"' . $since,
+			'ALL FROM "chad@pitchperfect.test" SUBJECT "festival"' . $since,
+		) === $criteria,
+		'comma-separated sender and subject filters expand to one search per combination'
+	);
+
+	$criteria = $call( $handler, 'buildSearchCriteria', array(
+		'search_criteria' => 'UNSEEN',
+		'subject_filter'  => 'say "hi"',
+	) );
+	$assert( 'UNSEEN SUBJECT "say \"hi\""' === $criteria, 'single subject stays a string and quotes are escaped' );
+
+	if ( $failures ) {
+		fwrite( STDERR, count( $failures ) . " failure(s)\n" );
+		exit( 1 );
+	}
+	echo "All email fetch filter checks passed.\n";
+}