<?php
/**
 * Migration engine: orchestrates the full migration workflow.
 *
 * @package Shiftora\Migration
 */

declare(strict_types=1);

namespace Shiftora\Migration;

use Shiftora\Api\Support\WriteFailure;
use Shiftora\Backup\BackupCleaner;
use Shiftora\Backup\BackupManager;
use Shiftora\Backup\RecoveryManager;
use Shiftora\Connection\ConnectionManager;
use Shiftora\Logging\MigrationLogger;
use Shiftora\Logging\ServerLog;
use Shiftora\Logging\TransferTrace;
use Shiftora\Migration\FindReplace\RuleSet;
use Shiftora\Migration\FindReplace\UrlRuleBuilder;
use Shiftora\Support\CacheFlusher;
use Shiftora\Support\DiagnosticsService;
use Shiftora\Support\ErrorReporter;
use Shiftora\Support\HttpClient;
use Shiftora\Support\HttpRequestException;
use Shiftora\Support\MigrationObservability;
use Shiftora\Support\RemoteErrorExplainer;
use Shiftora\Support\TableNameScoper;
use RuntimeException;

// phpcs:disable WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages are not output.

// Prevent direct file access.
if ( ! defined( 'ABSPATH' ) ) {
	exit;
}

/**
 * Orchestrates one migration: preflight, backup, a chunked export/import loop
 * with per-chunk find/replace, and completion. Stateless between chunks; all
 * progress lives in the database.
 */
final class MigrationEngine {

	/** Largest raw file (8 MiB) sent in a batch. Keep in sync with FileBatchExportEndpoint::BATCH_FILE_MAX. */
	private const BATCH_FILE_THRESHOLD = 8_388_608;

	/**
	 * Encoded-body ceiling (24 MB) per batch request. The receiver counts encoded
	 * bytes (4/3 of raw), so TransferSizer divides by 4/3 for the raw budget.
	 */
	private const BATCH_ENCODED_CEILING = 24_000_000;

	/** Maximum files per batch. Keep in sync with FileBatchExportEndpoint::BATCH_MAX_FILES. */
	private const BATCH_MAX_FILES = 250;

	/**
	 * Minimum age before a stage sweep may delete a scratch file. A write in
	 * progress has a fresh mtime, so age is the only safe filter.
	 */
	private const ABANDONED_SCRATCH_MIN_AGE_SECONDS = HOUR_IN_SECONDS;

	/** The source's authenticated table route, shared by the size report and its fallback. */
	private const EXPORT_TABLES_ROUTE = '/shiftora/v1/export/tables';

	/**
	 * Shown when a pull cannot learn the source's table set. Avoids the words
	 * "permission" and "token", which ErrorReporter rewrites into a different failure.
	 */
	private const SOURCE_TABLE_LIST_UNREADABLE = 'Shiftora could not read the list of tables from the source site, so it does not know which tables this migration should replace. Nothing has been transferred. Check that Shiftora is active and current on the source site and that the two sites can reach each other, then start the migration again.';

	/**
	 * When this worker's window closes, as a UNIX timestamp with a fraction. A float
	 * so guard_deadline() (time()) and can_wait() (microtime()) agree in the last second.
	 */
	private ?float $active_deadline = null;

	/** @var array<string, array<string, mixed>> Pull chunks already fetched in a batch, by cursor. */
	private array $pull_prefetch = [];

	private int $pull_prefetch_owner = 0;

	/** Only worth it when this engine lives for many chunks; a fresh engine per chunk would discard the batch. */
	private bool $batch_pulls = false;

	public function enable_pull_batching(): void {
		$this->batch_pulls = true;
	}

	/** Consecutive pull windows the source answered with a 5xx before the run is failed. */
	private const PULL_SERVER_ERROR_LIMIT = 6;

	/** Rows per page a pull stops shrinking at after server errors. */
	private const PULL_SHRINK_FLOOR_ROWS = 500;

	/** Per-file transfer progress store. */
	private readonly FileProgressTable $file_progress;

	/** What the last worker learned about this link, per migration and stage. */
	private readonly TransferSizingStore $transfer_sizing;

	public function __construct(
		private readonly \wpdb             $wpdb,
		private readonly ConnectionManager $connections,
		private readonly MigrationRequestAuthenticator $request_authenticator,
		private readonly FileOperationAuditor $file_auditor,
		private readonly BackupManager     $backup_manager,
		private readonly BackupCleaner     $backup_cleaner,
		private readonly DatabaseExporter  $exporter,
		private readonly DatabaseImporter  $importer,
		private readonly PreflightChecker  $preflight,
		private readonly HttpClient        $http,
		private readonly ?\Shiftora\Backup\RollbackManager $rollback_manager = null,
		private readonly ?RecoveryManager $recovery_manager = null,
		private readonly ?MigrationWorkerLock $worker_lock = null,
		private readonly ?TransferRegistry $transfer_registry = null,
		private readonly ?TransferOperationLock $transfer_operation_lock = null
	) {
		$this->file_progress      = new FileProgressTable( $this->wpdb );
		$this->transfer_sizing    = new TransferSizingStore( $this->wpdb );
	}

	/**
	 * Stop when the worker deadline has passed. Runs before fetch_with_retry, which
	 * rethrows DeadlineExceeded and so cannot swallow it.
	 *
	 * @throws DeadlineExceeded When the deadline is expired.
	 */
	private function guard_deadline(): void {
		if ( null !== $this->active_deadline && microtime( true ) >= $this->active_deadline ) {
			throw new DeadlineExceeded( 'Worker deadline reached; deferring to the next scheduled chunk.' );
		}
	}

	/** Matches how the stored cursor resumes: by key when the table has one, else by offset. */
	private static function pull_cursor_key( string $bare_table, int $offset, ?string $last_pk ): string {
		return null !== $last_pk ? "{$bare_table}|pk|{$last_pk}" : "{$bare_table}|off|{$offset}";
	}

	private function request_timeout(): int {
		return HttpClient::timeout_for_deadline( $this->active_deadline );
	}

	/**
	 * Raw bytes this batch may carry. Learned per migration and stage, because link
	 * capacity and the worker's remaining time change.
	 */
	private function batch_budget_bytes( int $migration_id, string $stage ): int {
		$sizer = $this->transfer_sizing->load( $migration_id, $stage, $this->batch_encoded_ceiling() );

		return $sizer->budget_for(
			$this->active_deadline,
			$this->request_timeout()
		);
	}

	/**
	 * Ceiling for one file batch's encoded bytes. Every sizer load must go through
	 * here: the ceiling also bounds how far a persisted budget may grow, so a load
	 * that passed the raw constant would silently clamp what another call raised.
	 *
	 * The sizer ramps toward the ceiling and backs off on failure. A response is held
	 * whole, json-decoded and base64-inflated by this worker, so it is bounded by the
	 * export side's response budget and this process's memory.
	 */
	private function batch_encoded_ceiling(): int {
		// Mirror of the export side's response shell arithmetic, so the bound is
		// derived from its budget rather than restated as a magic number.
		$response_cap = \Shiftora\Api\Endpoints\FileBatchExportEndpoint::BATCH_RESPONSE_MAX
			- ( self::BATCH_MAX_FILES * 192 ) - 512;
		$memory       = (int) wp_convert_hr_to_bytes( (string) ini_get( 'memory_limit' ) );
		// A third, because the response exists as the raw body, the decoded JSON and
		// the inflated files at once. Floor at the default so a small
		// memory_limit never drops below the proven path.
		$memory_cap = $memory > 0 ? max( self::BATCH_ENCODED_CEILING, intdiv( $memory, 3 ) ) : $response_cap;

		return max( TransferSizer::MIN_BUDGET_BYTES, min( self::BATCH_ENCODED_CEILING, $response_cap, $memory_cap ) );
	}

	/** Drop per-run state when a run ends: completion, failure or rolled-back failure. */
	private function release_run_state( int $migration_id ): void {
		// Release the transfer's credential with the transfer.
		$this->release_transfer_state( $migration_id );

		// Extensions holding per-run state drop it here, so a later run never inherits it.
		do_action( 'shiftora_migration_released', $migration_id );
	}

	/** Drop what the run learned about its link. Best effort: a failed delete must not change a finished migration's outcome. */
	private function release_transfer_state( int $migration_id ): void {
		try {
			$this->transfer_sizing->forget( $migration_id );
		} catch ( \Throwable $e ) {
			// Housekeeping must never decide a migration's fate.
			unset( $e );
		}
	}

	/**
	 * Persist what a batch taught so the next worker starts there. Best effort.
	 *
	 * @param bool $completed Only a run that transferred every file may lower the opening size, see TransferSizingStore::remember_settled().
	 */
	private function record_batch_sizing( int $migration_id, string $stage, TransferSizer $sizer, bool $completed = false ): void {
		try {
			$this->transfer_sizing->save( $migration_id, $stage, $sizer );
			$this->transfer_sizing->remember_settled( $stage, $sizer, $completed );
		} catch ( \Throwable $e ) {
			// Sizing is an optimization and must never decide a migration's fate.
			unset( $e );
		}
	}

	/**
	 * Record a failed batch, shrinking only when the failure was about size. A
	 * refusal for another reason (parse error, unowned transfer) looks like an
	 * oversize 400, and halving for it would drive the size to its floor.
	 *
	 * @param int|null $status HTTP status, or null/0 when no response arrived.
	 */
	private function record_batch_failure( int $migration_id, string $stage, TransferSizer $sizer, ?int $status ): void {
		$this->record_batch_sizing(
			$migration_id,
			$stage,
			FileBatchSizer::is_size_failure( $status ) ? $sizer->with_failure() : $sizer
		);
	}

	/**
	 * Initialise a migration: preflight, create the record, take the backup.
	 *
	 * @param array<string, mixed>                                                 $stages       Optional stages config, keyed by stage id.
	 * @param array<int, array{find: string, replace: string, generated: false}>   $custom_rules Validated rules the user typed, run after the URL rules.
	 *
	 * @throws RuntimeException On preflight failure or DB error.
	 */
	public function initialise(
		int $connection_id,
		MigrationDirection $direction,
		array $stages = [],
		bool $dry_run = false,
		array $custom_rules = []
	): MigrationRecord {
		$wpdb = $this->wpdb;

		$has_file_stages = $this->has_enabled_file_stages( $stages );

		try {
			$connection = $this->connections->find_by_id( $connection_id );

			// Preflight.
			$checks = $this->preflight->run( $connection_id );
			if ( ! PreflightChecker::all_passed( $checks ) ) {
				$failed_checks = array_filter( $checks, fn( $c ) => empty( $c['pass'] ) );
				$messages      = array_map( fn( $c ) => $c['message'], $failed_checks );
				throw new RuntimeException( 'Preflight failed: ' . implode( ' | ', $messages ) );
			}

			if ( ! $connection ) {
				throw new RuntimeException( 'Connection not found.' );
			}

			// Pull sizes come from the remote install so progress and ETA use its real
			// figures. The same call yields the source's table set; if that is unreadable the
			// run is refused before the record exists.
			$remote_size_report = $this->resolve_pull_remote_report( $connection, $this->database_stage_enabled( $stages ) );

			// The tables the request named, cut to the eligible set, or every eligible
			// table when it named none.
			$scope_rules = $this->resolve_scope( $direction, $stages, $remote_size_report );
			$tables      = $scope_rules['tables'];
			$stages      = $scope_rules['stages'];

			$tables = array_values( array_unique( array_filter( array_map(
				fn( $table ): string => $this->normalise_bare_table_name( (string) $table, $connection ),
				$tables
			) ) ) );

			$find_replace = $this->build_find_replace_rules( $connection, $custom_rules );

			// The destination tables don't exist yet and cannot be counted locally, so
			// total_rows stays 0 and progress is based on table completion.
			$total_rows = 0;
			$database_bytes_total = $this->estimate_database_stage_bytes( $tables, $stages, $remote_size_report );
			$stage_plan           = $this->build_stage_plan( $stages, $tables );
			$first_stage          = (string) ( $stage_plan[0]['key'] ?? '' );
			$starts_with_database = 'database' === $first_stage;

			// Prevent duplicate active migrations for the same connection and direction.
			// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
			$existing = $wpdb->get_row(
				$wpdb->prepare(
					"SELECT id, status FROM %i
					 WHERE connection_id = %d AND direction = %s
					   AND status IN ('running', 'pending', 'paused')
					 ORDER BY created_at DESC LIMIT 1",
					$wpdb->prefix . 'shiftora_migrations',
					$connection_id,
					$direction->value
				)
			);
			if ( $existing ) {
				if ( 'paused' === $existing->status ) {
					$this->resume( (int) $existing->id );
					return $this->load( (int) $existing->id );
				}
				throw new RuntimeException(
					__( 'A migration is already in progress for this connection.', 'shiftora' )
				);
			}

			// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
			$inserted = $wpdb->insert(
				$wpdb->prefix . 'shiftora_migrations',
				[
					'connection_id'      => $connection_id,
					'direction'          => $direction->value,
					'status'             => MigrationStatus::Pending->value,
					'tables'             => wp_json_encode( $tables ),
					'find_replace_rules' => ! empty( $find_replace ) ? wp_json_encode( $find_replace ) : null,
					'total_rows'         => $total_rows,
					'total_tables'       => count( $tables ),
					'processed_rows'     => 0,
					'processed_tables'   => 0,
						'tables_json'        => wp_json_encode( $tables ),
						'find_replace_json'  => ! empty( $find_replace ) ? wp_json_encode( $find_replace ) : null,
						'stages_json'        => ! empty( $stages ) ? wp_json_encode( $stages ) : null,
						'stage_plan_json'    => wp_json_encode( $stage_plan ),
						'completed_stages_json' => wp_json_encode( [] ),
						'dry_run'            => $dry_run ? 1 : 0,
						'error_message'      => null,
					'started_at'         => current_time( 'mysql', true ),
					'created_at'         => current_time( 'mysql', true ),
					'updated_at'         => current_time( 'mysql', true ),
						'current_stage'      => '' !== $first_stage ? $first_stage : null,
					'stage_progress'     => 0,
					'bytes_total'        => $starts_with_database ? $database_bytes_total : 0,
					'bytes_transferred'  => 0,
					'total_files'        => $starts_with_database ? count( $tables ) : 0,
					'transferred_files'  => 0,
				],
				[ '%d', '%s', '%s', '%s', '%s', '%d', '%d', '%d', '%d', '%s', '%s', '%s', '%s', '%s', '%d', '%s', '%s', '%s', '%s', '%s', '%d', '%d', '%d', '%d', '%d' ]
			);

			if ( ! $inserted ) {
				throw new RuntimeException( 'Failed to create migration record.' );
			}

			$migration_id = $wpdb->insert_id;

			$logger = ( new MigrationLogger( $migration_id ) )
				->with_trace( TransferTrace::sender( $migration_id, $direction->value ) );

			$logger->info( 'Migration initialised.', [
				'connection_id' => $connection_id,
				'direction'     => $direction->value,
				'tables'        => $tables,
				'total_rows'    => $total_rows,
				'file_only'     => empty( $tables ) && $has_file_stages,
			] );

			// Its old staging debris is swept here.
			if ( $this->transfer_registry ) {
				try {
					$this->transfer_registry->sweep_stale_staging();
				} catch ( \Throwable $e ) {
					$logger->warning( 'Could not sweep old staging tables: ' . $e->getMessage() );
				}
			}

			if ( ! empty( $scope_rules['left_out'] ) ) {
				$logger->info( TableEligibility::summarise( $scope_rules['left_out'] ), [
					'reasons' => array_values( array_unique( $scope_rules['left_out'] ) ),
				] );
			}

			$logger->info( sprintf(
				'Auto-generated %d URL find/replace rule(s): %s → %s.',
				count( $find_replace ),
				...$this->site_urls( $connection )
			) );

			// Snapshot before any DB-changing run.
			if ( ! $dry_run && ! empty( $tables ) ) {
				// Pull selections are stored as bare names; map them back to the local
				// prefix before generating the backup.
				$backup_tables = array_values( array_unique( array_map(
					fn( string $table ): string => str_starts_with( $table, (string) $wpdb->prefix )
						? $table
						: (string) $wpdb->prefix . $table,
					$tables
				) ) );
				$backup_tables = $this->backup_manager->prepare_backup_tables( $backup_tables );

				if ( ! empty( $backup_tables ) ) {
					try {
						if ( $this->rollback_manager ) {
							$record_for_rollback = $this->load( $migration_id );
							$backup = $this->rollback_manager->create_restore_point( $record_for_rollback );
						} else {
							$backup = $this->backup_manager->create( $migration_id, $backup_tables );
						}

						if ( $backup ) {
							$logger->info( 'Restore point created before migration.', [ 'backup_id' => $backup->id, 'size' => $backup->file_size ] );
							if ( $this->recovery_manager ) {
								$this->recovery_manager->create_lock( $migration_id, $backup->id );
							}
						}
						$this->backup_cleaner->clean();

						// Restore points have their own pool, kept to the last ten.
						if ( $this->rollback_manager ) {
							$cleaned = $this->rollback_manager->cleanup();
							if ( $cleaned > 0 ) {
								$logger->info( "Cleaned up {$cleaned} old rollback snapshot(s)." );
							}
						}
					} catch ( RuntimeException $e ) {
						$this->fail( $migration_id, 'Backup failed: ' . $e->getMessage() );
						throw $e;
					}
				} else {
					$logger->info( 'Skipped restore point because the selected tables do not exist on this destination yet.', [
						'direction' => $direction->value,
						'tables'    => $tables,
					] );
				}
			}

			$this->update_status( $migration_id, MigrationStatus::Running );
			$logger->info( 'Migration started.' );
			MigrationObservability::emit_migration_initialized(
				(string) $migration_id,
				$connection_id,
				$direction->value,
				count( $tables ),
				$database_bytes_total
			);

			// The run gets an identity before its first chunk: it writes receipts locally and
			// needs it up front. An existing identity is kept so a resume finds its receipts.
			if ( ! $dry_run && ! empty( $tables ) ) {
				try {
					$this->ensure_persisted_transfer_uuid( $migration_id );
				} catch ( RuntimeException $e ) {
					$this->fail( $migration_id, $e->getMessage() );
					throw $e;
				}
			}

			return $this->load( $migration_id );
		} catch ( RuntimeException $e ) {
			ServerLog::write( sprintf(
				'Shiftora initialise(): %s (connection=%d, direction=%s, file_stages=%s)',
				$e->getMessage(),
				$connection_id,
				$direction->value,
				$has_file_stages ? 'yes' : 'no'
			) );
			throw $e;
		}
	}

	/**
	 * Process the next chunk for a running migration. MigrationJob calls this until
	 * it returns false.
	 *
	 * @return bool True if there are more chunks to process.
	 *
	 * @throws RuntimeException On error.
	 */
	public function process_next_chunk( int $migration_id, ?float $deadline = null ): bool {
		$wpdb = $this->wpdb;

		if ( ! $this->acquire_migration_lock( $migration_id ) ) {
			return true;
		}

		$this->active_deadline = $deadline;

		try {
			$this->guard_deadline();

	$record = $this->load( $migration_id );

	// Bound here so every entry this loop writes, including retry warnings from
	// fetch_with_retry, carries the transfer it belongs to.
	$logger = ( new MigrationLogger( $migration_id ) )->with_trace(
		TransferTrace::sender( $migration_id, $record->direction->value )
			->with_uuid( $record->transfer_uuid )
	);

		if ( $record->status !== MigrationStatus::Running ) {
			return false;
		}

		if ( ! empty( $record->file_offsets_json ) && 'database' !== ( $record->current_stage ?? '' ) ) {
			return true;
		}

		$connection = $this->connections->find_by_id( $record->connection_id );
		if ( ! $connection ) {
			$this->fail( $migration_id, 'Connection not found.' );
			return false;
		}

		$tables   = $record->tables();
		$stages   = $record->stages();
		$rule_set = RuleSet::from_array( $record->find_replace_rules() );
		$dry_run  = $record->dry_run;

		// Offsets live on the migration record, not in transients, because wp_options
		// itself is truncated during import.
		$offsets = [];
		if ( ! empty( $record->table_offsets_json ) ) {
			$decoded = json_decode( $record->table_offsets_json, true );
			if ( is_array( $decoded ) ) {
				$offsets = $decoded;
			}
		}

		// Staged table names accumulate as staging tables are created.
		$staged_tables = [];
		if ( ! empty( $record->staged_tables_json ) ) {
			$decoded = json_decode( $record->staged_tables_json, true );
			if ( is_array( $decoded ) ) {
				$staged_tables = $decoded;
			}
		}

		if ( ! $dry_run && $this->should_prepare_file_stage_before_database( $record, $stages, $offsets ) ) {
				$first_file_stage = $this->next_enabled_file_stage( $stages, '', $record );
				if ( null !== $first_file_stage ) {
					$logger->info( sprintf( "Starting file stage '%s' before the database stage.", $first_file_stage ) );
					$this->prepare_file_stage_generic( $migration_id, $record, $first_file_stage );
				return true;
			}
		}

		foreach ( $tables as $table ) {
			$key   = md5( $table );
			$entry = $offsets[ $key ] ?? [ 'offset' => 0, 'done' => false ];

			if ( $entry['done'] ) {
				continue;
			}

			// Accept both numeric-offset entries and keyset entries.
			$offset          = (int) ( $entry['offset'] ?? 0 );
			$last_pk         = isset( $entry['last_pk'] ) ? ( '' !== $entry['last_pk'] ? (string) $entry['last_pk'] : null ) : null;
			$imported_table  = $this->normalise_bare_table_name( $table, $connection );
			$start_ns        = hrtime( true );

			// Fetch a chunk from the source, import into this site.
			$chunk_size = $this->resolve_database_chunk_size();
			$route_path = '/shiftora/v1/export/pull-sql';
			$bare_table = $this->normalise_bare_table_name( $table, $connection );

			static $chunk_logged = false;
			if ( $offset <= 1 && ! $chunk_logged ) {
				$chunk_logged = true;
				$logger->info( "Database pull chunk size: {$chunk_size} rows (SQL format)" );
			}

			$this->guard_deadline();

			$cursor_key = self::pull_cursor_key( $bare_table, $offset, $last_pk );
			if ( $this->pull_prefetch_owner !== $migration_id || ! isset( $this->pull_prefetch[ $cursor_key ] ) ) {
				$this->pull_prefetch       = [];
				$this->pull_prefetch_owner = $migration_id;

				// This table's cursor first, then every pending table after it.
				$cursors  = [ [ 'table' => $bare_table, 'offset' => $offset, 'last_pk' => $last_pk ] ];
				$position = $this->batch_pulls ? array_search( $table, $tables, true ) : false;
				foreach ( array_slice( $tables, false === $position ? count( $tables ) : $position + 1 ) as $next_table ) {
					$next = $offsets[ md5( $next_table ) ] ?? [ 'offset' => 0, 'done' => false ];
					if ( ! empty( $next['done'] ) ) {
						continue;
					}
					$cursors[] = [
						'table'   => $this->normalise_bare_table_name( $next_table, $connection ),
						'offset'  => (int) ( $next['offset'] ?? 0 ),
						'last_pk' => isset( $next['last_pk'] ) && '' !== $next['last_pk'] ? (string) $next['last_pk'] : null,
					];
				}

				// After a server error ask for less: one table, half the rows per failed window.
				[ $cursors, $chunk_size ] = $this->shrink_after_server_errors( $migration_id, $cursors, $chunk_size );

				$pull_body = [
					'table'        => $bare_table,
					'offset'       => $offset,
					'chunk_size'   => $chunk_size,
					'find_replace' => $record->find_replace_rules(),
				];
				if ( null !== $last_pk ) {
					$pull_body['last_pk'] = $last_pk;
				}
				$full_url = rtrim( $connection->remote_url, '/' ) . '/wp-json' . $route_path;

				$server_error = null;
				try {
					$response = $this->fetch_with_retry(
						function () use ( $connection, $route_path, $full_url, $pull_body, $cursors, &$server_error ) {
							$timeout = $this->request_timeout();
							if ( count( $cursors ) > 1 ) {
								// Leave half the timeout for the transfer itself; recomputed per attempt as time runs out.
								$pull_body['tables']    = $cursors;
								$pull_body['max_bytes'] = 4 * 1024 * 1024;
								$pull_body['max_ms']    = min( 10000, max( 500, $timeout * 500 ) );
							}
							$payload = (string) wp_json_encode( $pull_body );
							$headers = $this->request_authenticator->create_auth_headers_for_connection( $connection, 'POST', $route_path, $payload );

							try {
								return $this->http->post( $full_url, $payload, [
									'Content-Type' => 'application/json',
									...$headers,
								], $timeout );
							} catch ( HttpRequestException $e ) {
								$server_error = $e->http_status() >= 500 ? $e : $server_error;
								throw $e;
							}
						},
						3,
						$logger,
						'pull SQL'
					);
				} catch ( DeadlineExceeded | HttpRequestException $e ) {
					// A yield hides the status, so the 5xx seen inside the attempt is what counts.
					if ( null !== $server_error ) {
						$this->count_server_error_window( $migration_id, $server_error, $logger );
					}
					throw $e;
				}

				$decoded = json_decode( (string) $response['body'], true );
				if ( is_array( $decoded ) && is_array( $decoded['batch'] ?? null ) ) {
					foreach ( $decoded['batch'] as $item ) {
						if ( ! is_array( $item ) ) {
							continue;
						}
						$item_lp = $item['request_last_pk'] ?? null;
						$item_key = self::pull_cursor_key(
							(string) ( $item['request_table'] ?? '' ),
							(int) ( $item['request_offset'] ?? 0 ),
							null !== $item_lp && '' !== $item_lp ? (string) $item_lp : null
						);
						$this->pull_prefetch[ $item_key ] = $item;
					}
				} else {
					// A source on an older version answers with the one chunk asked for.
					$this->pull_prefetch[ $cursor_key ] = $decoded;
				}
			}

			$chunk = $this->pull_prefetch[ $cursor_key ] ?? null;
			unset( $this->pull_prefetch[ $cursor_key ] );
			if ( ! is_array( $chunk ) || ! isset( $chunk['sql_gz'], $chunk['has_more'], $chunk['page_identity'] ) ) {
				throw new RuntimeException( "Invalid SQL chunk response from remote for table {$table}." );
			}

			$elapsed_ms = (int) round( ( hrtime( true ) - $start_ns ) / 1_000_000 );
			$logger->debug( "SQL chunk fetched: {$bare_table}", [
				'offset'     => $offset,
				'rows'       => $chunk['rows_count'] ?? 0,
				'elapsed_ms' => $elapsed_ms,
			] );

			// Resolved before the import: the receipt is written inside the importer's
			// transaction, so a missing identity must stop the chunk.
			$transfer_uuid = $this->pull_transfer_identity( $record );

			// The source chose this name, and it reaches staging names, the finalize list
			// and listeners. Only the table that was asked for is accepted.
			TableNameScoper::assert_pulled( $bare_table, $this->normalise_bare_table_name( (string) ( $chunk['table'] ?? $bare_table ), $connection ) );

			$this->importer->import_sql_chunk(
				$chunk['sql_gz'],
				$bare_table,
				$chunk['schema'] ?? null,
				true,
				$dry_run,
				$migration_id,
				'',
				$transfer_uuid,
				(string) ( $chunk['page_identity'] ?? '' ),
				(string) ( $connection->remote_db_prefix ?? '' )
			);

			delete_transient( self::server_error_streak_key( $migration_id ) );

			$imported_table = $bare_table;
			if ( ! $dry_run && ! empty( $chunk['schema'] ) && ! in_array( $bare_table, $staged_tables, true ) ) {
				$staged_tables[] = $bare_table;
			}

			// Persist the cursor and bump processed_rows with SQL arithmetic to avoid a
			// stale-read overwrite race.
			$row_count        = (int) ( $chunk['rows_count'] ?? 0 );
			// Use the keyset cursor when available, else the numeric offset for tables without a primary key.
			$new_cursor       = isset( $chunk['last_pk'] ) && null !== $chunk['last_pk']
				? [ 'last_pk' => $chunk['last_pk'], 'done' => ! $chunk['has_more'] ]
				: [ 'offset' => $chunk['next_offset'], 'done' => ! $chunk['has_more'] ];
			$offsets[ $key ]  = $new_cursor;
			$chunk_elapsed_ms = (int) round( ( hrtime( true ) - $start_ns ) / 1_000_000 );

			// A pulled table is complete in staging. Listeners may inspect it before the
			// swap; one that throws stops the run like any other import failure.
			if ( ! $dry_run && ! $chunk['has_more'] ) {
				do_action(
					'shiftora_pull_table_imported',
					(int) $migration_id,
					'pull',
					$imported_table,
					$wpdb->prefix . $imported_table,
					$this->importer->staging_table_name( $imported_table, '' ),
					$this->active_deadline
				);
			}

		$now                = current_time( 'mysql', true );
		$table_completed    = ! $chunk['has_more'] ? 1 : 0;
		$completed_tables   = count( array_filter( $offsets, static fn( array $offset_entry ): bool => ! empty( $offset_entry['done'] ) ) );
		$new_processed      = $record->processed_rows + $row_count;
		$database_total     = count( $tables );
		$chunk_bytes        = $this->estimate_chunk_bytes( $chunk );
		$bytes_total        = $record->bytes_total > 0 ? $record->bytes_total : $this->estimate_database_stage_bytes( $tables, $stages );
		$bytes_transferred  = $record->bytes_transferred + $chunk_bytes;
		if ( $bytes_total > 0 ) {
			$bytes_transferred = min( $bytes_total, $bytes_transferred );
		}
		$db_progress        = $bytes_total > 0 && $bytes_transferred > 0
			? (int) floor( min( 100, $bytes_transferred / $bytes_total * 100 ) )
			: ( $record->total_rows > 0 && $new_processed > 0
				? (int) floor( min( 100, $new_processed / $record->total_rows * 100 ) )
				: ( $database_total > 0 ? (int) floor( min( 100, $completed_tables / $database_total * 100 ) ) : 0 ) );
		// Progress never decreases within a migration.
		if ( $db_progress < $record->stage_progress ) {
			$db_progress = $record->stage_progress;
		}
		$owns_row = $this->write_database_progress(
			$migration_id,
			[
				'rows'              => $row_count,
				'tables'            => $table_completed,
				'offsets'           => $offsets,
				'staged_tables'     => $staged_tables,
				'progress'          => $db_progress,
				'bytes_total'       => $bytes_total,
				'bytes_transferred' => $bytes_transferred,
				'total_tables'      => $database_total,
				'done_tables'       => $completed_tables,
			],
			$now
		);

		if ( ! $owns_row ) {
			// A cancel or pause won while the chunk was in flight. Further writes or the
			// completion commit would override that decision, so the worker stops here.
			$logger->warning(
				'Worker lost ownership mid-chunk; stopping without completing.',
				[ 'migration_id' => $migration_id ]
			);
			return false;
		}

			// Update the recovery lock with current progress.
			if ( $this->recovery_manager ) {
				$this->recovery_manager->update_progress( $migration_id, 'database', $db_progress );
			}

			MigrationObservability::emit_migration_progress(
				(string) $migration_id,
				'database',
				$completed_tables,
				$database_total,
				$bytes_transferred,
				$chunk_elapsed_ms
			);

			if ( $chunk_elapsed_ms > 5000 ) {
				MigrationObservability::emit_performance_warning(
					(string) $migration_id,
					'slow_operation',
					'database_chunk_duration_ms',
					$chunk_elapsed_ms,
					5000
				);
			}

			// Log when a table finishes.
			if ( ! $chunk['has_more'] ) {
				$logger->info( "Table complete: {$imported_table}", [
					'elapsed_ms' => $chunk_elapsed_ms,
				] );
			}

			// Check whether all work is now done.
			$has_more_chunks = $chunk['has_more'];
			$has_pending     = $this->has_pending_from_offsets( $offsets, $tables );

				if ( ! $has_more_chunks && ! $has_pending ) {
					$this->mark_stage_completed( $migration_id, 'database' );
					if ( $this->complete( $migration_id ) ) {
						$logger->info( 'Migration completed successfully.' );
					}
					return false;
			}

			return $has_more_chunks || $has_pending;
		}

			// All tables were already done at the start of this invocation.
			$this->mark_stage_completed( $migration_id, 'database' );
			if ( $this->complete( $migration_id ) ) {
				$logger->info( 'Migration completed successfully.' );
			}
			return false;
		} finally {
			$this->active_deadline = null;
			$this->release_migration_lock( $migration_id );
		}
	}

	/** Mark a migration failed and roll back any staging tables it created. */
		public function fail( int $migration_id, string|\Throwable|\WP_Error $error ): void {
		$record          = null;
		$normalized      = null;
			$display_message = '';

		// Roll back staging tables so the real tables remain intact.
		try {
			$record = $this->load( $migration_id );

			// An operator's cancel is the last word. A worker that was mid-request when it
			// landed must not turn the canceled run into a failed one; cancel has already
			// done the cleanup.
			if ( MigrationStatus::Cancelled === $record->status ) {
				( new MigrationLogger( $migration_id ) )->info(
					'A failure arrived after the migration was canceled; it is logged and the run stays canceled.',
					[ 'error' => $error instanceof \Throwable ? $error->getMessage() : ( $error instanceof \WP_Error ? $error->get_error_message() : (string) $error ) ]
				);
				return;
			}

			if ( $record->direction === MigrationDirection::Pull ) {
				$this->rollback_pull_staging( $record );
			}
		} catch ( RuntimeException $inner ) {
			// The record may be corrupt; carry on with fail.
		}

		$error = $this->out_of_space_failure( $migration_id, $record, $error );
		$error = $this->readable_remote_failure( $migration_id, $error );

		$reporter    = new ErrorReporter();
		$diagnostics = new DiagnosticsService( $reporter );
		$normalized  = $reporter->normalize(
			$error,
			[
				'migration_id' => $migration_id,
				'stage'        => $record?->current_stage ?? 'migration',
			]
		);
		$display_message = (string) ( $normalized['user_message'] ?? (string) ( $normalized['message'] ?? 'Migration failed. Check the migration logs for details.' ) );

		// Mark the recovery lock so RecoveryManager can offer auto-recovery.
		if ( $this->recovery_manager ) {
			$this->recovery_manager->mark_failed( $migration_id, $display_message );
		}

		// Store the raw message so the status endpoint's re-normalization can match
		// ErrorReporter patterns. The client displays error_details.user_message.
		$stored_error_message = (string) ( $normalized['message'] ?? $normalized['user_message'] ?? 'Migration failed. Check the migration logs for details.' );

		if ( $record && empty( $record->tables() ) ) {
			$stored_error_message .= ' No database restore point was needed or available for this file-only migration.';
		}

		$this->wpdb->update(
			$this->wpdb->prefix . 'shiftora_migrations',
			[
				'status'        => MigrationStatus::Failed->value,
				'error_message' => $stored_error_message,
				'updated_at'    => current_time( 'mysql', true ),
			],
			[ 'id' => $migration_id ],
			[ '%s', '%s', '%s' ],
			[ '%d' ]
		);

		$this->release_run_state( $migration_id );

		$diagnostics->record_error( $normalized );
		$technical_message = (string) ( $normalized['technical_message'] ?? $normalized['message'] ?? $display_message );
		( new MigrationLogger( $migration_id ) )->error(
			$technical_message,
			[
				'normalized_code' => $normalized['code'] ?? 'unknown_error',
				'severity'        => $normalized['severity'] ?? 'error',
				'user_message'    => $display_message,
				'debug'           => $normalized['debug'] ?? [],
			]
		);

		MigrationObservability::emit_migration_error(
			(string) $migration_id,
			(string) ( $normalized['code'] ?? 'migration_failed' ),
			$display_message,
			[
				'stage' => (string) ( $record?->current_stage ?? 'migration' ),
			]
		);

		if ( $diagnostics->auto_report_enabled() && 'critical' === ( $normalized['severity'] ?? '' ) ) {
			try {
				$diagnostics->report_issue(
					[
						'summary'           => 'Automatic Shiftora critical failure report',
						'current_operation' => (string) ( $record?->current_stage ?? 'migration' ),
						'error'             => $display_message,
						'migration_id'      => $migration_id,
					]
				);
			} catch ( \Throwable $ignored ) {
				// Reporting must never break the primary failure flow.
			}
		}
	}

	/**
	 * Swaps raw transport text (a status line, a cURL error, a page the other site sent)
	 * for the readable explanation, and keeps the original line in the migration log.
	 * Any other error passes through.
	 *
	 * @param string|\Throwable|\WP_Error $error What failed.
	 */
	private function readable_remote_failure( int $migration_id, string|\Throwable|\WP_Error $error ): string|\Throwable|\WP_Error {
		if ( $error instanceof \WP_Error ) {
			return $error;
		}

		$line = $error instanceof \Throwable
			? RemoteErrorExplainer::line_for( $error )
			: RemoteErrorExplainer::line_for_text( $error );

		if ( null === $line ) {
			return $error;
		}

		( new MigrationLogger( $migration_id ) )->error(
			'The other site could not be reached or refused a request: ' . ( $error instanceof \Throwable ? $error->getMessage() : $error )
		);

		return $line;
	}

	/**
	 * Swaps a database-full error for the message an operator can act on, and keeps
	 * the original line in the log. Any other error passes through.
	 *
	 * @param string|\Throwable|\WP_Error $error What failed.
	 */
	private function out_of_space_failure( int $migration_id, ?MigrationRecord $record, string|\Throwable|\WP_Error $error ): string|\Throwable|\WP_Error {
		$raw = $error instanceof \Throwable
			? $error->getMessage()
			: ( $error instanceof \WP_Error ? $error->get_error_message() : $error );

		if ( ! OutOfSpace::matches( $raw ) ) {
			return $error;
		}

		( new MigrationLogger( $migration_id ) )->error( 'The destination reported it is out of space: ' . $raw );

		return OutOfSpace::message( $this->database_bytes_needed( $record, $error ) );
	}

	/**
	 * Bytes of table data this run moves, as the source reported when the run started.
	 * Zero when unknown.
	 */
	private function database_bytes_needed( ?MigrationRecord $record, mixed $error ): int {
		if ( $error instanceof DestinationOutOfSpace && $error->bytes_needed > 0 ) {
			return $error->bytes_needed;
		}

		return null === $record ? 0 : (int) $record->bytes_total;
	}

	/**
	 * Drop the staging tables a pull left behind, under the destination lock. Shared
	 * by fail() and cancel().
	 *
	 * Skipped when another migration owns the lease or the advisory lock is
	 * contended: rolling back would clobber its commit, so the owner recovers it.
	 */
	private function rollback_pull_staging( MigrationRecord $record ): void {
		$staged = json_decode( $record->staged_tables_json ?? '[]', true );
		if ( ! is_array( $staged ) || empty( $staged ) ) {
			return;
		}

		$scope  = 'dest:' . $record->connection_id;
		$locked = ! $this->transfer_operation_lock
			|| $this->transfer_operation_lock->acquire( $scope, TransferOperationLock::DEFAULT_ACQUIRE_TIMEOUT );

		try {
			if ( ! $locked ) {
				( new MigrationLogger( $record->id ) )->warning( 'Rollback skipped: another session holds the destination operation lock.' );
			} elseif ( $this->transfer_registry
				&& $this->transfer_registry->destination_held_by_other( $record->connection_id, $record->id ) ) {
				( new MigrationLogger( $record->id ) )->warning( 'Rollback skipped: another migration owns the destination connection lease.' );
			} else {
				// Scoped to this transfer's identity so the purge cannot take the receipts of
				// another transfer that reused this migration id.
				$this->importer->rollback_staging(
					$staged,
					$record->id,
					'',
					$this->pull_transfer_identity( $record )
				);
			}
		} finally {
			if ( $this->transfer_operation_lock && $locked ) {
				$this->transfer_operation_lock->release( $scope );
			}
		}
	}

	/** Cancel a running or paused migration and drop the staging tables its pull left behind. */
	public function cancel( int $migration_id ): void {
		$logger = new MigrationLogger( $migration_id );
		$record = $this->load( $migration_id );

		$applied = $this->transition_status(
			$migration_id,
			[ MigrationStatus::Running, MigrationStatus::Pending, MigrationStatus::Paused ],
			MigrationStatus::Cancelled
		);

		if ( ! $applied ) {
			$logger->info( 'Cancel ignored, the migration already reached a terminal state.' );
			MigrationJob::cancel( $migration_id );
			return;
		}

		// Unscheduling is always safe: a committed migration has no pending jobs, and
		// a live one must not hand on to the next chunk.
		MigrationJob::cancel( $migration_id );

		// fail() does not run for a canceled migration, so the run cleans up here.
		try {
			$this->rollback_pull_staging( $record );
		} catch ( RuntimeException $e ) {
			$logger->warning( 'Cancelled, but cleaning up after the run failed: ' . $e->getMessage() );
		}

		if ( $this->recovery_manager ) {
			$this->recovery_manager->clear_lock( $migration_id );
		}
		$this->release_run_state( $migration_id );
		$logger->info( 'Migration canceled.' );
	}

	/**
	 * Ends a cancel that no worker is left to finish.
	 *
	 * Whatever is still cancelling is forced to cancelled and cleaned up the way
	 * cancel() does, best effort. The status guard means a worker that finishes in
	 * the meantime wins.
	 */
	public function finish_stuck_cancel( int $migration_id ): void {
		$logger = new MigrationLogger( $migration_id );
		$record = $this->load( $migration_id );

		if ( MigrationStatus::Cancelling !== $record->status ) {
			return;
		}

		if ( ! $this->transition_status( $migration_id, [ MigrationStatus::Cancelling ], MigrationStatus::Cancelled ) ) {
			return;
		}

		try {
			$this->rollback_pull_staging( $record );
		} catch ( RuntimeException $e ) {
			$logger->warning( 'Cancelled, but cleaning up after the run failed: ' . $e->getMessage() );
		}

		if ( $this->recovery_manager ) {
			$this->recovery_manager->clear_lock( $migration_id );
		}
		$this->release_run_state( $migration_id );
		$logger->info( 'Migration canceled; no worker was left to finish the cancel.' );
	}

	/**
	 * Pause a running or pending migration and cancel its scheduled jobs. DB data is untouched.
	 *
	 * @throws RuntimeException If the migration is not in a pauseable state.
	 */
	public function pause( int $migration_id ): void {
		$record = $this->load( $migration_id );

		if ( ! in_array( $record->status, [ MigrationStatus::Running, MigrationStatus::Pending ], true ) ) {
			throw new RuntimeException(
				"Migration #{$migration_id} cannot be paused (status: {$record->status->value})."
			);
		}

		MigrationJob::cancel( $migration_id );

		$applied = $this->transition_status(
			$migration_id,
			[ MigrationStatus::Running, MigrationStatus::Pending ],
			MigrationStatus::Paused
		);

		( new MigrationLogger( $migration_id ) )->info(
			$applied
				? 'Migration paused.'
				: 'Pause skipped, a concurrent change already moved the migration on.'
		);
	}

	/**
	 * Pause when the background scheduler is unavailable.
	 *
	 * Unlike pause(), this keeps all state and scheduled jobs and sets error_message
	 * so the UI offers "Resume" instead of "Failed".
	 */
	public function pause_for_scheduler_unavailable( int $migration_id, string $message ): void {
		$record = $this->load( $migration_id );

		if ( ! in_array( $record->status, [ MigrationStatus::Running, MigrationStatus::Pending ], true ) ) {
			return;
		}

		// Same fence as pause(): only a running row may be parked, and the ownership
		// claim is revoked in the same statement.
		$applied = $this->transition_status(
			$migration_id,
			[ MigrationStatus::Running, MigrationStatus::Pending ],
			MigrationStatus::Paused
		);

		if ( ! $applied ) {
			return;
		}

		$this->wpdb->update(
			$this->wpdb->prefix . 'shiftora_migrations',
			[ 'error_message' => $message, 'updated_at' => current_time( 'mysql', true ) ],
			[ 'id' => $migration_id ],
			[ '%s', '%s' ],
			[ '%d' ]
		);

		( new MigrationLogger( $migration_id ) )->warning(
			'Migration paused because background worker scheduling is unavailable. Resume is safe after scheduler is active.',
			[ 'message' => $message ]
		);
	}

	/**
	 * Resume a paused migration by re-scheduling its next chunk.
	 *
	 * @throws RuntimeException If the migration is not paused.
	 */
	public function resume( int $migration_id ): void {
		$record = $this->load( $migration_id );

		if ( $record->status !== MigrationStatus::Paused ) {
			throw new RuntimeException(
				"Migration #{$migration_id} cannot be resumed (status: {$record->status->value})."
			);
		}

		$scheduled = ! empty( $record->file_offsets_json )
			? MigrationJob::schedule_file( $migration_id )
			: MigrationJob::schedule( $migration_id );

		if ( ! $scheduled ) {
			throw new RuntimeException(
				__( 'Migration could not resume: the background task scheduler is unavailable.', 'shiftora' )
			);
		}

		// Marking Running only after the job is queued means a failed schedule leaves
		// the migration paused. A cancel that wins the race in between is not overridden.
		$applied = $this->transition_status(
			$migration_id,
			[ MigrationStatus::Paused ],
			MigrationStatus::Running
		);

		if ( ! $applied ) {
			MigrationJob::cancel( $migration_id );
			throw new RuntimeException(
				__( 'Migration could not resume: another change moved it on first.', 'shiftora' )
			);
		}

		( new MigrationLogger( $migration_id ) )->info( 'Migration resumed.' );
	}

	/**
	 * Load a MigrationRecord by ID.
	 *
	 * @throws RuntimeException If not found.
	 */
	public function load( int $id ): MigrationRecord {
		$wpdb = $this->wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$row = $wpdb->get_row(
			$wpdb->prepare(
				'SELECT * FROM %i WHERE id = %d',
				$wpdb->prefix . 'shiftora_migrations',
				$id
			)
		);

		if ( ! $row ) {
			throw new RuntimeException( "Migration #{$id} not found." );
		}

		return MigrationRecord::from_row( $row );
	}

	// Private helpers.

	private function update_status( int $id, MigrationStatus $status ): void {
		$this->wpdb->update(
			$this->wpdb->prefix . 'shiftora_migrations',
			[ 'status' => $status->value, 'updated_at' => current_time( 'mysql', true ) ],
			[ 'id' => $id ],
			[ '%s', '%s' ],
			[ '%d' ]
		);
	}

	/**
	 * Compare-and-swap a migration into $to, but only from one of $allowed.
	 *
	 * The single UPDATE also clears the worker lock columns, so a stranded worker's
	 * next guarded write sees 0 rows. Returns false when a concurrent action already
	 * moved the row.
	 *
	 * @param MigrationStatus[] $allowed States the migration may currently be in.
	 */
	private function transition_status( int $id, array $allowed, MigrationStatus $to ): bool {
		$wpdb = $this->wpdb;

		$allowed_values = array_values( array_map(
			static fn( MigrationStatus $s ): string => $s->value,
			$allowed
		) );
		$now   = current_time( 'mysql', true );
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$affected = (int) $wpdb->query( $wpdb->prepare(
			'UPDATE %i
			    SET status                = %s,
			        worker_lock_owner     = NULL,
			        worker_lock_expires   = NULL,
			        updated_at            = %s
			  WHERE id = %d
			    AND status IN (' . implode( ', ', array_fill( 0, count( $allowed_values ), '%s' ) ) . ')',
			array_merge( [ $wpdb->prefix . 'shiftora_migrations', $to->value, $now, $id ], $allowed_values )
		) );

		// Only when the transition landed: a cancel that found nothing to cancel has
		// not ended a transfer and must not clear one that is still running.
		if ( $affected > 0 && $to->is_terminal() ) {
			$this->release_transfer_state( $id );
		}

		return $affected > 0;
	}

	/**
	 * Adds a chunk's counts to the migration row, but only while this worker still owns it.
	 *
	 * @param array{rows: int, tables: int, offsets: array<mixed>, staged_tables: array<mixed>, progress: int, bytes_total: int, bytes_transferred: int, total_tables: int, done_tables: int} $progress
	 *
	 * @return bool False when a cancel or pause took the row while the chunk ran.
	 */
	private function write_database_progress( int $migration_id, array $progress, string $now ): bool {
		$wpdb = $this->wpdb;

		[ $owner_checked, $owner_token ] = $this->ownership_guard();

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		return (bool) $wpdb->query( $wpdb->prepare(
			"UPDATE %i
			    SET processed_rows     = processed_rows + %d,
			        processed_tables   = processed_tables + %d,
			        table_offsets_json = %s,
			        staged_tables_json = %s,
			        current_stage      = 'database',
			        stage_progress     = %d,
			        bytes_total        = %d,
			        bytes_transferred  = %d,
			        total_files        = %d,
			        transferred_files  = %d,
			        last_heartbeat_at  = %s,
			        updated_at         = %s
			  WHERE id = %d AND status = 'running' AND (%d = 0 OR worker_lock_owner = %s)",
			$wpdb->prefix . 'shiftora_migrations',
			$progress['rows'],
			$progress['tables'],
			wp_json_encode( $progress['offsets'] ),
			wp_json_encode( $progress['staged_tables'] ),
			$progress['progress'],
			$progress['bytes_total'],
			$progress['bytes_transferred'],
			$progress['total_tables'],
			$progress['done_tables'],
			$now,
			$now,
			$migration_id,
			$owner_checked,
			$owner_token
		) );
	}

	/**
	 * WHERE guard that only lets the current running owner write the row. Without a
	 * lock the status check alone fences the write.
	 *
	 * @return array{0: int, 1: string} Whether the lock owner is checked, and the owner token to compare.
	 */
	private function ownership_guard(): array {
		$token = $this->worker_lock?->token();

		return [ null === $token ? 0 : 1, (string) $token ];
	}

	/**
	 * Re-read the row before a commit boundary and decide whether this worker still
	 * owns it. A stop or resume revokes the lock in one UPDATE, so committing could
	 * apply tables an owner canceled. Falls back to the status fence when no claim is held.
	 */
	private function commit_boundary_owned( int $id ): bool {
		$wpdb = $this->wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$row = $wpdb->get_row(
			$wpdb->prepare(
				'SELECT status, worker_lock_owner FROM %i WHERE id = %d',
				$wpdb->prefix . 'shiftora_migrations',
				$id
			),
			ARRAY_A
		);
		if ( ! is_array( $row ) || 'running' !== (string) ( $row['status'] ?? '' ) ) {
			return false;
		}

		$token = $this->worker_lock?->token();
		if ( null === $token ) {
			return true;
		}
		return $token === (string) ( $row['worker_lock_owner'] ?? '' );
	}

	/**
	 * Persist the transfer's uuid so a resumed run keeps the same identity. A lost
	 * write fails the migration.
	 *
	 * @throws RuntimeException When the row cannot be updated.
	 */
	private function record_transfer_uuid( int $id, string $transfer_uuid ): void {
		$updated = $this->wpdb->update(
			$this->wpdb->prefix . 'shiftora_migrations',
			[ 'transfer_uuid' => $transfer_uuid, 'updated_at' => current_time( 'mysql', true ) ],
			[ 'id' => $id ],
			[ '%s', '%s' ],
			[ '%d' ]
		);

		if ( false === $updated ) {
			throw new RuntimeException(
				sprintf( 'Failed to record the transfer UUID for migration %d: %s', $id, (string) $this->wpdb->last_error )
			);
		}
	}

	/**
	 * Reuse the persisted identity, or mint and persist one before the first chunk so a
	 * retried start keeps the original.
	 *
	 * @throws RuntimeException When the persisted value is not a uuid, or the row cannot be stamped.
	 */
	private function ensure_persisted_transfer_uuid( int $id ): string {
		$record = $this->load( $id );

		if ( is_string( $record->transfer_uuid ) && preg_match( '/^[0-9a-f]{32}$/', $record->transfer_uuid ) ) {
			return $record->transfer_uuid;
		}

		if ( null !== $record->transfer_uuid && '' !== $record->transfer_uuid ) {
			throw new RuntimeException( sprintf(
				'Persisted transfer uuid for migration %d is malformed: refusing to mint a competing identity.',
				$id
			) );
		}

		$uuid = bin2hex( random_bytes( 16 ) );
		$this->record_transfer_uuid( $id, $uuid );

		return $uuid;
	}

	/**
	 * The identity a pull's receipts are written and read under.
	 *
	 * Receipts are scoped by (migration_id, transfer_uuid, chunk_key); ids restart
	 * after a reset, so the id alone is not enough. A row without an identity predates
	 * stamping: writing would share the empty-string namespace with legacy receipts,
	 * so this refuses and the operator restarts. Reads of a legacy record still resolve to ''.
	 *
	 * @throws RuntimeException When the record carries no usable identity.
	 */
	private function pull_transfer_identity( MigrationRecord $record ): string {
		$uuid = is_string( $record->transfer_uuid ) ? $record->transfer_uuid : '';

		if ( ! preg_match( '/^[0-9a-f]{32}$/', $uuid ) ) {
			throw new RuntimeException( sprintf(
				/* translators: %d: migration id */
				__(
					'Migration %d has no transfer identity, so its chunk receipts cannot be scoped. Start the migration again to continue.',
					'shiftora'
				),
				$record->id
			) );
		}

		return $uuid;
	}

	/**
	 * Build the durable stage plan; file stages precede database.
	 *
	 * @return array<int, array<string, mixed>>
	 */
	private function build_stage_plan( array $stages, array $tables ): array {
		$plan = [];
		foreach ( FileStages::keys() as $stage_key ) {
			if ( ! empty( $stages[ $stage_key ]['enabled'] ) ) {
				$plan[] = [
					'key'          => $stage_key,
					'status'       => 'pending',
					'started_at'   => null,
					'completed_at' => null,
				];
			}
		}

		if ( ! empty( $tables ) && $this->database_stage_enabled( $stages ) ) {
			$plan[] = [
				'key'          => 'database',
				'status'       => 'pending',
				'started_at'   => null,
				'completed_at' => null,
			];
		}

		return $plan;
	}

	/**
	 * @return array<int, array<string, mixed>>
	 */
	private function stage_plan_for_record( MigrationRecord $record ): array {
		$plan = $record->stage_plan();
		return ! empty( $plan ) ? $plan : $this->build_stage_plan( $record->stages(), $record->tables() );
	}

	/**
	 * @return array<string, mixed>
	 */
	private function completed_stages_for_record( MigrationRecord $record ): array {
		return $record->completed_stages();
	}

	private function stage_is_completed( MigrationRecord $record, string $stage_key ): bool {
		if ( '' === $stage_key ) {
			return false;
		}

		$completed = $this->completed_stages_for_record( $record );
		if ( isset( $completed[ $stage_key ] ) ) {
			return true;
		}

		foreach ( $this->stage_plan_for_record( $record ) as $entry ) {
			if ( $stage_key === (string) ( $entry['key'] ?? '' ) && 'completed' === (string) ( $entry['status'] ?? '' ) ) {
				return true;
			}
		}

		return false;
	}

	private function mark_stage_completed( int $id, string $stage_key ): void {
		$wpdb = $this->wpdb;

		$record = $this->load( $id );
		if ( $this->stage_is_completed( $record, $stage_key ) ) {
			return;
		}

		$now       = current_time( 'mysql', true );
		$completed = $this->completed_stages_for_record( $record );
		$completed[ $stage_key ] = [ 'completed_at' => $now ];
		$plan = $this->stage_plan_with_status( $this->stage_plan_for_record( $record ), $stage_key, 'completed', null, $now );

		// The database stage forces bytes_transferred and transferred_files up to their
		// totals so the status response never shows a finished stage as partial. Any
		// other stage leaves both as they are.
		$is_database       = 'database' === $stage_key;
		$bytes_transferred = max( max( 0, (int) $record->bytes_total ), (int) $record->bytes_transferred );
		$files_transferred = max( max( 0, (int) $record->total_files ), (int) $record->transferred_files );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$wpdb->query( $wpdb->prepare(
			'UPDATE %i
			    SET stage_plan_json       = %s,
			        completed_stages_json = %s,
			        stage_progress        = 100,
			        updated_at            = %s,
			        bytes_transferred     = IF(%d, %d, bytes_transferred),
			        transferred_files     = IF(%d, %d, transferred_files)
			  WHERE id = %d',
			$wpdb->prefix . 'shiftora_migrations',
			wp_json_encode( $plan ),
			wp_json_encode( $completed ),
			$now,
			$is_database ? 1 : 0,
			$bytes_transferred,
			$is_database ? 1 : 0,
			$files_transferred,
			$id
		) );
	}

	/**
	 * @param array<int, array<string, mixed>> $plan
	 * @return array<int, array<string, mixed>>
	 */
	private function stage_plan_with_status( array $plan, string $stage_key, string $status, ?string $started_at, ?string $completed_at ): array {
		$found = false;
		foreach ( $plan as &$entry ) {
			if ( $stage_key !== (string) ( $entry['key'] ?? '' ) ) {
				continue;
			}

			$found = true;
			$entry['status'] = $status;
			if ( null !== $started_at && empty( $entry['started_at'] ) ) {
				$entry['started_at'] = $started_at;
			}
			if ( null !== $completed_at ) {
				$entry['completed_at'] = $completed_at;
			}
		}
		unset( $entry );

		if ( ! $found ) {
			$plan[] = [
				'key'          => $stage_key,
				'status'       => $status,
				'started_at'   => $started_at,
				'completed_at' => $completed_at,
			];
		}

		return $plan;
	}

	/** Whether any non-database file stage is enabled. */
	private function has_enabled_file_stages( array $stages ): bool {
		foreach ( FileStages::keys() as $stage_key ) {
			if ( ! empty( $stages[ $stage_key ]['enabled'] ) ) {
				return true;
			}
		}

		return false;
	}

	/**
	 * Acquire the shared migration worker lock.
	 */
	private function acquire_migration_lock( int $migration_id ): bool {
		if ( null === $this->worker_lock ) {
			return true;
		}

		return $this->worker_lock->claim( $migration_id );
	}

	/**
	 * Release the shared migration worker lock.
	 */
	private function release_migration_lock( int $migration_id ): void {
		$this->worker_lock?->release( $migration_id );
	}

	/** The next enabled file stage after the given stage key. */
	private function next_enabled_file_stage( array $stages, string $after_stage = '', ?MigrationRecord $record = null ): ?string {
		$ordered    = FileStages::keys();
		$start_from = 0;

		if ( '' !== $after_stage ) {
			$current_index = array_search( $after_stage, $ordered, true );
			if ( false !== $current_index ) {
				$start_from = (int) $current_index + 1;
			}
		}

		for ( $i = $start_from; $i < count( $ordered ); $i++ ) {
			if ( ! empty( $stages[ $ordered[ $i ] ]['enabled'] )
				&& ( null === $record || ! $this->stage_is_completed( $record, $ordered[ $i ] ) ) ) {
				return $ordered[ $i ];
			}
		}

		return null;
	}

	/** Whether the database stage still has pending work. */
	private function database_stage_has_pending_work( MigrationRecord $record, array $offsets = [] ): bool {
		$tables = $record->tables();
		$stages = $record->stages();

		if ( empty( $tables ) ) {
			return false;
		}

		if ( $this->stage_is_completed( $record, 'database' ) ) {
			return false;
		}

		if ( ! $this->database_stage_enabled( $stages ) ) {
			return false;
		}

		if ( empty( $offsets ) && ! empty( $record->table_offsets_json ) ) {
			$decoded = json_decode( $record->table_offsets_json, true );
			if ( is_array( $decoded ) ) {
				$offsets = $decoded;
			}
		}

		return empty( $offsets ) || $this->has_pending_from_offsets( $offsets, $tables );
	}

	/** Start file stages before the database when both are selected. */
	private function should_prepare_file_stage_before_database( MigrationRecord $record, array $stages, array $offsets ): bool {
		if ( ! $this->has_enabled_file_stages( $stages ) ) {
			return false;
		}

		if ( ! empty( $record->file_offsets_json ) || 'database' === ( $record->current_stage ?? '' ) ) {
			return false;
		}

		return $record->processed_rows <= 0 && empty( $offsets );
	}

	/**
	 * After the last file stage completes, hand off to the database stage when enabled.
	 */
	private function handoff_to_database_stage( int $id, MigrationRecord $record, MigrationLogger $logger ): bool {
		if ( ! $this->database_stage_has_pending_work( $record ) ) {
			return false;
		}

		$now                  = current_time( 'mysql', true );
		$database_bytes_total = $this->estimate_database_stage_bytes( $record->tables(), $record->stages() );
		$plan                 = $this->stage_plan_with_status(
			$this->stage_plan_for_record( $record ),
			'database',
			'running',
			$now,
			null
		);

		$this->wpdb->update(
			$this->wpdb->prefix . 'shiftora_migrations',
			[
				'file_offsets_json'  => null,
				'current_stage'      => 'database',
				'stage_progress'     => 0,
				'stage_plan_json'    => wp_json_encode( $plan ),
				'bytes_total'        => $database_bytes_total,
				'bytes_transferred'  => 0,
				'total_files'        => count( $record->tables() ),
				'transferred_files'  => 0,
				'last_heartbeat_at'  => $now,
				'updated_at'         => $now,
			],
			[ 'id' => $id ],
			[ '%s', '%s', '%d', '%s', '%d', '%d', '%d', '%d', '%s', '%s' ],
			[ '%d' ]
		);

		$logger->info( 'File stages complete. Starting database stage last.' );

		MigrationJob::schedule( $id );

		return true;
	}

	/**
	 * Normalize a table name to a bare WordPress table identifier.
	 */
	private function normalise_bare_table_name( string $table_name, object $connection ): string {
		$normalized    = $table_name;
		$local_prefix  = (string) $this->wpdb->prefix;
		$remote_prefix = (string) ( $connection->remote_db_prefix ?? '' );

		foreach ( array_unique( array_filter( [ $local_prefix, $remote_prefix ] ) ) as $prefix ) {
			if ( '' !== $prefix && str_starts_with( $normalized, $prefix ) ) {
				$normalized = substr( $normalized, strlen( $prefix ) );
				break;
			}
		}

		return $normalized;
	}

	/**
	 * Effective database chunk size. A stored legacy 500 becomes the current default
	 * (10000); custom values of 1000 or more are kept, capped at MAX_CHUNK_SIZE.
	 *
	 * @return int Rows per chunk (1000-25000).
	 */
	private function resolve_database_chunk_size(): int {
		$raw = (int) get_option( 'shiftora_chunk_size', DatabaseExporter::DEFAULT_CHUNK_SIZE );

		return DatabaseExporter::effective_chunk_size( $raw );
	}

	/**
	 * Estimate the selected database size for the current stage. Trusts the source's size
	 * report from initialise; the client-supplied value is only a fallback for paths that
	 * skip it.
	 *
	 * @param array<string, mixed>|null $remote_report Remote /export/tables payload.
	 */
	private function estimate_database_stage_bytes( array $tables, array $stages = [], ?array $remote_report = null ): int {
		if ( empty( $tables ) ) {
			return 0;
		}

		if ( is_array( $remote_report ) ) {
			return max( 0, (int) ( $remote_report['total_size_bytes'] ?? 0 ) );
		}

		return max( 0, (int) ( $stages['database']['size_bytes'] ?? 0 ) );
	}

	/**
	 * Ask the remote for authoritative table and file sizes, so a pull's progress
	 * total is the remote's figure, not the wizard's estimate.
	 *
	 * @return array<string, mixed>|null Null when the report is unavailable.
	 */
	private function fetch_remote_size_report( object $connection ): ?array {
		$query_string = '?' . http_build_query( [
			'include_uploads_size' => '1',
			'include_scannables'   => '1',
		] );

		return $this->request_export_tables( $connection, $query_string );
	}

	/**
	 * Ask for the table list alone, without the file scan. The scan is the half that
	 * runs out of time or memory on a large site, so a source that cannot produce it
	 * can still say what tables it has.
	 *
	 * @return array<string, mixed>|null Null when the list is unavailable.
	 */
	private function fetch_remote_table_list( object $connection ): ?array {
		return $this->request_export_tables( $connection, '' );
	}

	/**
	 * One authenticated call to the source's table route.
	 *
	 * @param string $query_string Query to append, or an empty string for none.
	 *
	 * @return array<string, mixed>|null Null on any transport or decode failure.
	 */
	private function request_export_tables( object $connection, string $query_string ): ?array {
		$full_url = rtrim( (string) ( $connection->remote_url ?? '' ), '/' ) . '/wp-json' . self::EXPORT_TABLES_ROUTE . $query_string;

		try {
			$this->guard_deadline();
			$headers  = $this->request_authenticator->create_auth_headers_for_connection( $connection, 'GET', self::EXPORT_TABLES_ROUTE . $query_string );
			$response = $this->http->get( $full_url, $headers, $this->request_timeout() );
		} catch ( \Throwable $e ) {
			return null;
		}

		$data = json_decode( (string) ( $response['body'] ?? '' ), true );
		if ( JSON_ERROR_NONE !== json_last_error() || ! is_array( $data ) ) {
			return null;
		}

		return $data;
	}

	/**
	 * Whether a payload carries the source's answer about its tables. `{}` decodes to
	 * an empty array, and reading that as "no tables" would replace nothing. The key
	 * must be present; present but empty means the source has none.
	 *
	 * @param array<string, mixed>|null $report Decoded remote payload.
	 *
	 * @return array<string, mixed>|null The payload when it answered, else null.
	 */
	private static function answered_with_tables( ?array $report ): ?array {
		return isset( $report['tables'] ) && is_array( $report['tables'] ) ? $report : null;
	}

	/**
	 * The find/replace rules this transfer applies, URL rules first: the source's URLs
	 * become this site's.
	 *
	 * @param array<int, array<string, mixed>> $custom_rules Rules the user typed.
	 *
	 * @return array<int, array<string, string>>
	 */
	private function build_find_replace_rules( object $connection, array $custom_rules ): array {
		[ $from_url, $to_url ] = $this->site_urls( $connection );

		return array_merge(
			UrlRuleBuilder::build( $from_url, $to_url ),
			$this->resolve_custom_find_replace_rules( $custom_rules )
		);
	}

	/**
	 * The two site URLs a rewrite goes between: the source's, then this site's. Shared
	 * with the log line that reports it so the two cannot disagree.
	 *
	 * @return array{0: string, 1: string} [ from, to ].
	 */
	private function site_urls( object $connection ): array {
		$remote = untrailingslashit( (string) ( $connection->remote_site_url ?? $connection->remote_url ?? '' ) );

		return [ $remote, untrailingslashit( get_site_url() ) ];
	}

	/**
	 * Establish what the source has, or refuse to start.
	 *
	 * A pull builds its database stage from the source's table list, so an unreadable
	 * list cannot pass as an empty one. Asks for the report, falls back to the bare
	 * table list, and stops the run when neither arrived and one is required. It has
	 * to stop here: later the destination would already be truncated.
	 *
	 * @param bool $required Whether this run cannot proceed without it.
	 *
	 * @return array<string, mixed>|null Null only when not required and unread.
	 *
	 * @throws RuntimeException When the source's table set is required and unread.
	 */
	private function resolve_pull_remote_report( object $connection, bool $required ): ?array {
		$report = self::answered_with_tables( $this->fetch_remote_size_report( $connection ) )
			?? self::answered_with_tables( $this->fetch_remote_table_list( $connection ) );

		if ( null !== $report || ! $required ) {
			return $report;
		}

		// Named for what failed, not a ceiling: the source would not answer, and the
		// operator's next step is on the source site.
		throw new RuntimeException( self::SOURCE_TABLE_LIST_UNREADABLE );
	}

	/**
	 * The scope is the eligible tables the request named, or every eligible table when
	 * it named none, plus every enabled file stage in whole. The list passes through
	 * `shiftora_migration_tables` once after that, and whatever comes back is screened
	 * again, so eligibility holds for any list.
	 *
	 * @param array<string, mixed> $stages Client-supplied stage configuration.
	 *
	 * @return array{ tables: string[], stages: array<string, mixed>, left_out: array<string, string> } `left_out` maps each table dropped as not this install's to the reason.
	 *
	 * @throws RuntimeException When the source's table list is unreadable, or the request named tables and none of them exist.
	 */
	private function resolve_scope( MigrationDirection $direction, array $stages, ?array $remote_report ): array {
		// A run with the database stage disabled asked for no tables, so widening
		// would hand the pipeline a table list for a migration with no database work.
		if ( ! $this->database_stage_enabled( $stages ) ) {
			return [ 'tables' => [], 'stages' => $stages, 'left_out' => [] ];
		}

		$requested = $this->requested_tables( $stages );

		// A pull without a readable list has zero tables, which a completed run would
		// report as its replacement. initialize() refuses this run first, so reaching
		// here means a caller skipped that check.
		$answered = self::answered_with_tables( $remote_report );
		if ( null === $answered ) {
			throw new RuntimeException( self::SOURCE_TABLE_LIST_UNREADABLE );
		}

		$tables = array_values( array_filter( array_unique( array_map(
			'strval',
			array_column( $answered['tables'], 'full_name' )
		) ) ) );

		// The remote filters its own list, but an older build does not, so it is screened
		// again here with the remote's prefix.
		$remote_prefix = self::remote_prefix( $remote_report );
		$eligible      = TableEligibility::screen_remote( $tables, $remote_prefix );
		$picked        = $this->pick_requested( $requested, $eligible['kept'], $remote_prefix );
		$carried       = $this->filter_migration_tables( $picked, $direction, $remote_prefix );

		// Only what the source advertised and is eligible: a name the filter made up is not on its side.
		return [
			'tables'   => array_values( array_intersect( $carried, $eligible['kept'] ) ),
			'stages'   => $stages,
			'left_out' => $eligible['left_out'],
		];
	}

	/**
	 * The table names the start request put on the database stage, full or bare.
	 * Empty means it named none, which is every eligible table.
	 *
	 * @param array<string, mixed> $stages Client-supplied stage configuration.
	 *
	 * @return string[]
	 */
	private function requested_tables( array $stages ): array {
		$named = $stages['database']['tables'] ?? [];
		if ( ! is_array( $named ) ) {
			return [];
		}

		return array_values( array_unique( array_filter( array_map(
			static fn( $name ): string => is_scalar( $name ) ? trim( (string) $name ) : '',
			$named
		) ) ) );
	}

	/**
	 * Narrows the eligible tables to the requested ones, in the eligible set's order.
	 * A name that is not in the set is dropped without comment: the wizard offers only
	 * names from that set, so one that is not there was typed by hand or has since gone.
	 *
	 * @param string[] $requested Names from the request, full or bare.
	 * @param string[] $eligible  Full names this run may carry.
	 * @param string   $prefix    Prefix those names carry.
	 *
	 * @return string[] Full names.
	 *
	 * @throws RuntimeException When names were requested and none of them is eligible.
	 */
	private function pick_requested( array $requested, array $eligible, string $prefix ): array {
		if ( [] === $requested ) {
			return $eligible;
		}

		$wanted = array_flip( $requested );
		$picked = array_values( array_filter(
			$eligible,
			static function ( string $full ) use ( $wanted, $prefix ): bool {
				$bare = '' !== $prefix && str_starts_with( $full, $prefix ) ? substr( $full, strlen( $prefix ) ) : $full;

				return isset( $wanted[ $full ] ) || isset( $wanted[ $bare ] );
			}
		) );

		if ( [] === $picked ) {
			throw new RuntimeException( __( 'None of the tables you picked exist on the site they are read from.', 'shiftora' ) );
		}

		return $picked;
	}

	/**
	 * @param string[] $tables Full names of the tables this run would carry.
	 * @param string   $prefix The source's prefix, which the names carry.
	 *
	 * @return string[]
	 */
	private function filter_migration_tables( array $tables, MigrationDirection $direction, string $prefix ): array {
		/**
		 * Filters the tables a migration carries. The list is already limited to what
		 * is eligible, and what comes back is screened again, so names this install
		 * cannot send are dropped whatever the filter returns.
		 *
		 * @param string[] $tables    Full table names.
		 * @param string   $direction Always 'pull'.
		 * @param string   $prefix    Database prefix the names carry.
		 */
		$carried = apply_filters( 'shiftora_migration_tables', $tables, $direction->value, $prefix );

		return is_array( $carried ) ? array_values( array_map( 'strval', $carried ) ) : $tables;
	}

	/**
	 * The remote site's table prefix from its report, or an empty string when it did not say.
	 *
	 * @param array<string, mixed>|null $report Decoded remote payload.
	 */
	private static function remote_prefix( ?array $report ): string {
		return null === $report ? '' : TableNaming::from_remote( $report )->base_prefix;
	}

	/**
	 * Whether this run includes a database stage. Shared by build_stage_plan() and
	 * database_stage_has_pending_work() so the table list and the plan describe the
	 * same run. An absent `database` key means enabled.
	 */
	private function database_stage_enabled( array $stages ): bool {
		return empty( $stages )
			|| ! isset( $stages['database'] )
			|| ! empty( $stages['database']['enabled'] );
	}

	/**
	 * Rules merged after the automatic URL rules: the ones the user typed, then
	 * whatever a developer adds through the filter. An add-on that read the same
	 * request returns the typed rules again, so a rule is kept once.
	 *
	 * @param array<int, array<string, mixed>> $typed Validated rules from the start request.
	 *
	 * @return array<int, array<string, mixed>> Rules for the merge.
	 */
	private function resolve_custom_find_replace_rules( array $typed ): array {
		/**
		 * Filters the extra find/replace rules a migration applies after the
		 * automatic URL rules. Each rule is an array with `find` and `replace`.
		 *
		 * @param array<int, array<string, mixed>> $rules The rules typed on the Scope step.
		 */
		$accepted = apply_filters( 'shiftora_custom_find_replace_rules', $typed );
		$accepted = is_array( $accepted ) ? array_values( array_filter( $accepted, 'is_array' ) ) : $typed;

		$seen   = [];
		$unique = [];
		foreach ( array_merge( $typed, $accepted ) as $rule ) {
			$key = wp_json_encode( [ $rule['find'] ?? '', $rule['replace'] ?? '' ] );
			if ( isset( $seen[ $key ] ) ) {
				continue;
			}
			$seen[ $key ] = true;
			$unique[]     = $rule;
		}

		return $unique;
	}

	private function estimate_chunk_bytes( array $chunk ): int {
		if ( isset( $chunk['chunk_bytes'] ) ) {
			return max( 0, (int) $chunk['chunk_bytes'] );
		}

		$payload = [
			'table'   => $chunk['table'] ?? '',
			'columns' => $chunk['columns'] ?? [],
			'rows'    => $chunk['rows'] ?? [],
			'schema'  => $chunk['schema'] ?? null,
		];

		return strlen( (string) ( wp_json_encode( $payload ) ?: '' ) );
	}

	private function complete( int $id ): bool {
		// Finalize staging tables for pull migrations.
		$record = $this->load( $id );
		if ( ! $record->dry_run && $record->direction === MigrationDirection::Pull ) {
			$staged = json_decode( $record->staged_tables_json ?? '[]', true );
			if ( is_array( $staged ) && ! empty( $staged ) ) {
				// The worker reaching the rename must still be the running owner. A cancel
				// that landed during the last chunk ends the run here instead of swapping tables
				// into a canceled row.
				if ( ! $this->commit_boundary_owned( $id ) ) {
					( new MigrationLogger( $id ) )->warning( 'Complete skipped: lost the worker lock before the PULL commit.' );
					return false;
				}

				// The rename is a server-side fence with one session owning the destination at
				// a time. Another session on this destination must not race it, so a
				// contended lock aborts the run the same way a lost lease does.
				$scope = 'dest:' . $record->connection_id;
				$locked = ! $this->transfer_operation_lock
					|| $this->transfer_operation_lock->acquire( $scope, TransferOperationLock::DEFAULT_ACQUIRE_TIMEOUT );

				try {
					if ( ! $locked ) {
						$this->fail( $id, __( 'PULL finalize refused: another session holds the destination operation lock.', 'shiftora' ) );
						return false;
					}

					// The destination is coordinated through the transfer lease: another pull
					// mid-flight holds the slot and wins the rename race.
					if ( $this->transfer_registry && $this->transfer_registry->destination_held_by_other( $record->connection_id, $id ) ) {
						$this->fail( $id, __( 'PULL finalize refused: another migration currently owns the destination connection lease.', 'shiftora' ) );
						return false;
					}
					$connection = $this->connections->find_by_id( $record->connection_id );
					$src_prefix = $connection->remote_db_prefix ?? '';
					// The commit marker is the durable proof this transfer replaced the tables, so
					// it carries the identity that makes it this transfer's proof and not the next one's.
					$this->importer->finalize_staging(
						$staged,
						$id,
						$src_prefix,
						'',
						null,
						null,
						$this->pull_transfer_identity( $record )
					);

					// The operator's session went with the old users table; their page may now reclaim one.
					SessionHandoff::mark_swapped( $id );

					// A pull from a foreign prefix imports _user_roles/_capabilities keys bound to
					// that prefix. Flag the one-shot repair so the next page load rewrites them.
					if ( '' !== $src_prefix && $src_prefix !== $this->wpdb->prefix ) {
						update_option( 'shiftora_repair_prefix_pending', true );
					}
				} finally {
					if ( $this->transfer_operation_lock && $locked ) {
						$this->transfer_operation_lock->release( $scope );
					}
				}
			}
			// Force-set siteurl and home to the local URLs after every pull. This catches
			// URL variants find/replace missed (http vs https, www prefix) that would
			// otherwise break the frontend. Each option is mapped independently because a
			// site can have a distinct home, such as siteurl without www and home with it.
			$site_url = untrailingslashit( get_site_url() );
			$home_url = untrailingslashit( get_home_url() );
			if ( '' !== $site_url || '' !== $home_url ) {
				$this->wpdb->update(
					$this->wpdb->prefix . 'options',
					[ 'option_value' => $site_url ],
					[ 'option_name'  => 'siteurl' ],
					[ '%s' ],
					[ '%s' ]
				);
				$this->wpdb->update(
					$this->wpdb->prefix . 'options',
					[ 'option_value' => $home_url ],
					[ 'option_name'  => 'home' ],
					[ '%s' ],
					[ '%s' ]
				);
				( new MigrationLogger( $id ) )->info( "Post-import: siteurl forced to local URL: {$site_url}" );
				( new MigrationLogger( $id ) )->info( "Post-import: home forced to local URL: {$home_url}" );
			}
		}

		if ( $record->dry_run && $this->has_enabled_file_stages( $record->stages() ) ) {
			foreach ( FileStages::keys() as $stage_key ) {
				if ( empty( $record->stages()[ $stage_key ]['enabled'] ) ) {
					continue;
				}

				$this->simulate_file_stage_generic( $id, $record, $stage_key );

				// A stage that could not be simulated fails the run here; continuing would walk
				// the remaining stages against a run already marked failed.
				if ( ! in_array( $this->load( $id )->status, [ MigrationStatus::Running, MigrationStatus::Pending ], true ) ) {
					break;
				}
			}
		}

		// Skipped when a stage above already ended the run: completing a failed run
		// asks the completion gate why its stages never finished, and that answer would
		// replace the one saying what went wrong.
		if ( ! in_array( $this->load( $id )->status, [ MigrationStatus::Running, MigrationStatus::Pending ], true ) ) {
			return false;
		}

		$this->complete_migration( $id );

		// Only clean up and clear the lock if the migration actually completed.
		$record_after_complete = $this->load( $id );
		if ( $record_after_complete->status !== MigrationStatus::Completed ) {
			( new MigrationLogger( $id ) )->warning( 'Complete: migration did not reach completed status, skipping cleanup and lock clear.' );
			return false;
		}

		// The operator's choice goes with the run; the results stay.
		$this->release_run_state( $id );

		// Clean up shiftora_old_* tables once completion is confirmed.
		if ( $record_after_complete->direction === MigrationDirection::Pull ) {
			// Another operation may have taken over the destination since this run
			// completed. Its commit owns the old tables now, so dropping them would destroy
			// the checkpoint it may still roll back to.
			if ( $this->transfer_registry && $this->transfer_registry->destination_held_by_other( $record_after_complete->connection_id, $id ) ) {
				( new MigrationLogger( $id ) )->warning( 'Old-table cleanup skipped: another migration owns the destination connection lease.' );
			} else {
				$old_tables = json_decode( $record->staged_tables_json ?? '[]', true );
				if ( is_array( $old_tables ) && ! empty( $old_tables ) ) {
					$this->importer->drop_old_tables( $old_tables );
				}
			}
		}

		// Clear the recovery lock only after complete_migration() succeeds.
		if ( $this->recovery_manager ) {
			$this->recovery_manager->clear_lock( $id );
		}

		return true;
	}

	// Generic file-stage pipeline (uploads, plus any stage an extension registers).

	/**
	 * Passes a stage's manifest through `shiftora_stage_manifest`, then orders it.
	 *
	 * @param array<int, array{path: string, size: int, hash: string}> $manifest
	 *
	 * @return array<int, array{path: string, size: int, hash: string}>
	 */
	private function scope_stage_manifest( array $manifest, string $stage_key, MigrationRecord $record ): array {
		/**
		 * Filters the files a file stage carries: the manifest a pull received, before any
		 * file is requested. A callback returns the entries to keep, or throws a
		 * RuntimeException to refuse the run. It has to be repeatable, since the same
		 * manifest is rebuilt when a pull heals drift.
		 *
		 * @param array<int, array{path: string, size: int, hash: string}> $manifest Entries with a relative `path`.
		 * @param string          $stage_key Stage key: uploads, or a stage added through `shiftora_file_stages`.
		 * @param MigrationRecord $record    The migration being run.
		 */
		$scoped = apply_filters( 'shiftora_stage_manifest', $manifest, $stage_key, $record );

		return $this->without_refused_files( self::only_listed( $manifest, $scoped ), $stage_key, $record );
	}

	/**
	 * What a filter returned, limited to entries of the original manifest. A filter can
	 * leave files out; it cannot add one or change what the source said about it.
	 *
	 * @param array<int, array{path: string, size: int, hash: string}> $manifest
	 * @param mixed                                                    $scoped   The filter's answer.
	 *
	 * @return array<int, array{path: string, size: int, hash: string}>
	 */
	private static function only_listed( array $manifest, mixed $scoped ): array {
		if ( ! is_array( $scoped ) ) {
			return $manifest;
		}

		$listed = [];
		foreach ( $manifest as $entry ) {
			$listed[ (string) ( $entry['path'] ?? '' ) ] = $entry;
		}

		$kept = [];
		foreach ( $scoped as $entry ) {
			$path = is_array( $entry ) ? (string) ( $entry['path'] ?? '' ) : '';
			if ( isset( $listed[ $path ] ) ) {
				$kept[ $path ] = $listed[ $path ];
			}
		}

		return array_values( $kept );
	}

	/**
	 * Drops the files Free's own writer would refuse, and says which. A file left
	 * out silently would look like a transfer that lost it.
	 *
	 * @param array<int, array{path: string, size: int, hash: string}> $manifest
	 *
	 * @return array<int, array{path: string, size: int, hash: string}>
	 */
	private function without_refused_files( array $manifest, string $stage_key, MigrationRecord $record ): array {
		$kept    = [];
		$refused = [];

		foreach ( $manifest as $entry ) {
			$path = (string) ( $entry['path'] ?? '' );

			if ( null === FileStages::refusal( $stage_key, $path ) ) {
				$kept[] = $entry;
			} else {
				$refused[] = $path;
			}
		}

		if ( [] !== $refused ) {
			( new MigrationLogger( (int) $record->id ) )->warning(
				sprintf(
					'Stage "%s": %d file(s) were skipped because their names are never written on the destination.',
					$stage_key,
					count( $refused )
				),
				[ 'skipped' => (string) wp_json_encode( array_slice( $refused, 0, 20 ) ) ]
			);
		}

		return $kept;
	}

	/**
	 * Convert a file manifest into tracked offset state.
	 *
	 * @param array<int, array{path: string, size: int, hash: string}> $manifest
	 *
	 * @return array{bytes_total: int, offsets: array<string, array<string, mixed>>}
	 */
	private function build_file_offsets_state( array $manifest ): array {
		$bytes_total = 0;
		$offsets     = [];

		foreach ( $manifest as $entry ) {
			$path = (string) $entry['path'];
			if ( '' === $path ) {
				continue;
			}

			$size             = (int) ( $entry['size'] ?? 0 );
			$hash             = (string) ( $entry['hash'] ?? '' );
			$bytes_total     += $size;
			$offsets[ $path ] = [
				'size'   => $size,
				'hash'   => $hash,
				'offset' => 0,
				'done'   => false,
			];
		}

		return [
			'bytes_total' => $bytes_total,
			'offsets'     => $offsets,
		];
	}

	/**
	 * Persist a file stage's manifest and offsets, reset its progress counters and
	 * record current_stage so process_next_file_chunk() resolves the right base directory.
	 *
	 * @param array<int, array{path: string, size: int, hash: string}> $manifest
	 * @param array<string, array<string, mixed>>                       $offsets
	 */
	private function persist_file_stage( int $id, string $stage_key, array $manifest, int $bytes_total, array $offsets ): void {
		$record = $this->load( $id );
		$now    = current_time( 'mysql', true );
		$plan   = $this->stage_plan_with_status(
			$this->stage_plan_for_record( $record ),
			$stage_key,
			'running',
			$now,
			null
		);

		$this->wpdb->update(
			$this->wpdb->prefix . 'shiftora_migrations',
			[
				'total_files'       => count( $manifest ),
				'bytes_total'       => $bytes_total,
				'transferred_files' => 0,
				'bytes_transferred' => 0,
				'file_offsets_json' => wp_json_encode( $offsets ),
				'current_stage'     => $stage_key,
				'stage_plan_json'   => wp_json_encode( $plan ),
				'stage_progress'    => 0,
				'updated_at'        => $now,
			],
			[ 'id' => $id ],
			[ '%d', '%d', '%d', '%d', '%s', '%s', '%s', '%d', '%s' ],
			[ '%d' ]
		);

		// Bring the per-file table into line with the manifest just built. The
		// counters above are zeroed, so offered files go back to pending and files the
		// manifest no longer lists are dropped, not left as pending work the remote
		// would refuse with a 403.
		$reconciled = $this->file_progress->reconcile_stage( $id, $stage_key, $offsets, false );

		if ( $reconciled['dropped'] > 0 ) {
			( new MigrationLogger( $id ) )->warning(
				sprintf(
					'Stage "%s": dropped %d progress row(s) the manifest no longer offers.',
					$stage_key,
					$reconciled['dropped']
				),
				[ 'migration_id' => $id, 'stage' => $stage_key ]
			);
		}
	}

	/**
	 * Bring a running stage into line with a refreshed manifest, keeping progress.
	 *
	 * persist_file_stage() resets every file to pending, which is wrong mid-stage: a
	 * dropped file would re-send everything delivered, and only uploads is
	 * deduplicated. Here only stale rows go, new paths are added as pending, and the rest
	 * keep their progress.
	 */
	private function refresh_file_stage( int $id, string $stage_key, array $manifest, int $bytes_total, array $offsets ): void {
		$this->wpdb->update(
			$this->wpdb->prefix . 'shiftora_migrations',
			[
				'total_files'       => count( $manifest ),
				'bytes_total'       => $bytes_total,
				'file_offsets_json' => wp_json_encode( $offsets ),
				'updated_at'        => current_time( 'mysql', true ),
			],
			[ 'id' => $id ],
			[ '%d', '%d', '%s', '%s' ],
			[ '%d' ]
		);

		$reconciled = $this->file_progress->reconcile_stage( $id, $stage_key, $offsets, true );

		if ( $reconciled['dropped'] > 0 || $reconciled['written'] > 0 ) {
			( new MigrationLogger( $id ) )->warning(
				sprintf(
					'Stage "%s": dropped %d row(s) the source no longer offers and added %d new one(s); progress on the rest is kept.',
					$stage_key,
					$reconciled['dropped'],
					$reconciled['written']
				),
				[ 'migration_id' => $id, 'stage' => $stage_key ]
			);
		}
	}

	/** Dry-run for a single file stage: enumerate what would be transferred and log a summary. */
	private function simulate_file_stage_generic( int $id, MigrationRecord $record, string $stage_key ): void {
		$logger = new MigrationLogger( $id );

		// A pull's dry run fetches the manifest for real. It is a signed read that
		// changes nothing, and it is the only way to tell the operator what the pull
		// would move, which for a narrowed stage is the whole question.
		try {
			$connection = $this->connections->find_by_id( $record->connection_id );
			if ( ! $connection ) {
				$logger->warning( sprintf( 'DRY RUN: connection #%d not found, could not ask the source.', $record->connection_id ) );
				return;
			}

			$manifest = $this->fetch_pull_manifest( $connection, $record, $stage_key, $logger );

			if ( null === $manifest ) {
				$logger->info( sprintf( 'DRY RUN: Stage "%s" has no files on the source, would skip.', $stage_key ) );
				return;
			}

			$total_size = (int) array_sum( array_column( $manifest, 'size' ) );

			$logger->info( sprintf(
				'DRY RUN: Stage "%s" would fetch %d file(s) from the source, totalling %s bytes.',
				$stage_key,
				count( $manifest ),
				number_format( $total_size )
			) );
		} catch ( DeadlineExceeded $e ) {
			throw $e;
		} catch ( HttpRequestException $e ) {
			// A refusal is not a warning. The stage was asked for by name and the source
			// does not have it, so a success report would claim a migration is fine when
			// it would be refused for real.
			if ( ! $e->is_retryable() ) {
				// Failed rather than thrown: this runs inside the request that started the
				// migration, and an exception here would orphan the run in `running` with a
				// worker queued against a stage that never starts.
				$this->fail(
					$id,
					sprintf(
						'Dry run cannot answer stage "%s": %s',
						$stage_key,
						$e->getMessage()
					)
				);
				return;
			}

			$logger->warning( sprintf( 'DRY RUN: could not ask the source about stage "%s", %s', $stage_key, $e->getMessage() ) );
		} catch ( RuntimeException $e ) {
			$logger->warning( sprintf( 'DRY RUN: could not ask the source about stage "%s", %s', $stage_key, $e->getMessage() ) );
		}

		// Marked complete so the run can finish. This is bookkeeping on the run's own
		// row and nothing was written or moved. Without it the completion gate sees a
		// stage nobody marked and reports the dry run as a failure.
		$this->mark_stage_completed( $id, $stage_key );
	}

	/**
	 * Prepare and schedule a file stage: fetch the source's manifest, persist it and
	 * enqueue the first chunk.
	 * With no files the stage is skipped and the next one advances.
	 *
	 * @param string $stage_key A registered file stage: uploads, or one added through `shiftora_file_stages`.
	 */
	private function prepare_file_stage_generic( int $id, MigrationRecord $record, string $stage_key ): void {
		$logger          = new MigrationLogger( $id );
		$record          = $this->load( $id );
		if ( $this->stage_is_completed( $record, $stage_key ) ) {
			$logger->info( "File stage '{$stage_key}' already completed, skipping permanently." );
			$this->advance_to_next_file_stage( $id, $logger, $stage_key, false );
			return;
		}

		$record = $this->load( $id );

		// Fetch the manifest from the source.
		$connection = $this->connections->find_by_id( $record->connection_id );
		if ( ! $connection ) {
			$logger->error( "File stage '{$stage_key}': connection not found, failing migration." );
			$this->fail( $id, "File stage '{$stage_key}' connection not found." );
			return;
		}

		try {
			$manifest = $this->fetch_pull_manifest( $connection, $record, $stage_key, $logger );
		} catch ( DeadlineExceeded $e ) {
			throw $e;
		} catch ( RuntimeException $e ) {
			$logger->error( "Failed to fetch manifest for stage '{$stage_key}': {$e->getMessage()}" );
			$this->fail( $id, "Failed to fetch file manifest for stage '{$stage_key}': {$e->getMessage()}" );
			return;
		}

		if ( null === $manifest ) {
			$logger->info( "File stage '{$stage_key}': no files in manifest, skipping (empty_manifest)." );
			$this->advance_to_next_file_stage( $id, $logger, $stage_key );
			return;
		}

		$file_state  = $this->build_file_offsets_state( $manifest );
		$bytes_total = $file_state['bytes_total'];
		$offsets     = $file_state['offsets'];

		$this->persist_file_stage( $id, $stage_key, $manifest, $bytes_total, $offsets );
		$logger->info( sprintf( "Pull file stage \"%s\": %d file(s), %d bytes.", $stage_key, count( $manifest ), $bytes_total ) );
		if ( ! $this->enqueue_file_chunk_job( $id, $logger ) ) {
			$this->fail( $id, "File stage '{$stage_key}' could not queue background worker." );
			return;
		}
	}

	/**
	 * Fetch a pull stage's file manifest from the remote. Shared with the drift-heal
	 * path so both ask the same route and read the answer the same way. The source
	 * lists the whole stage; the stage manifest filter is applied before any file is
	 * requested.
	 *
	 * @return array<string, array<string, mixed>>|null Manifest, or null when there are no files to fetch (not an error).
	 *
	 * @throws RuntimeException When the manifest could not be obtained or parsed.
	 */
	private function fetch_pull_manifest( object $connection, MigrationRecord $record, string $stage_key, MigrationLogger $logger ): ?array {
		$route_path   = '/shiftora/v1/export/media-manifest';
		$query_string = '?' . http_build_query( [ 'stage' => $stage_key ] );
		$full_url     = rtrim( $connection->remote_url, '/' ) . '/wp-json' . $route_path . $query_string;

		$this->guard_deadline();
		$response = $this->fetch_with_retry(
			function () use ( $connection, $route_path, $query_string, $full_url ) {
				// The query string already carries its own '?', and the receiver rebuilds the
				// signed path from the route plus the raw QUERY_STRING. Signing
				// route . '?' . $query_string would cover "/manifest??stage=uploads", which is
				// not the request sent, so the signature would never match.
				$headers = $this->request_authenticator->create_auth_headers_for_connection( $connection, 'GET', $route_path . $query_string );

				return $this->http->get( $full_url, [
					...$headers,
				], $this->request_timeout() );
			},
			3,
			$logger,
			"media manifest"
		);

		$data = json_decode( $response['body'], true );
		if ( JSON_ERROR_NONE !== json_last_error() ) {
			throw new RuntimeException( "File stage '{$stage_key}' manifest JSON is corrupt." );
		}
		if ( ! is_array( $data ) ) {
			throw new RuntimeException( "File stage '{$stage_key}' manifest response is invalid." );
		}

		// "No files" is a manifest that says so, and that is the only shape meaning an
		// empty stage. Anything else that is not a usable file list is a source that
		// answered with something unexpected. Treating it as empty would tick the stage
		// off with nothing transferred.
		if ( ! array_key_exists( 'files', $data ) || ! is_array( $data['files'] ) ) {
			throw new RuntimeException( "File stage '{$stage_key}' manifest response carries no usable file list." );
		}

		foreach ( $data['files'] as $entry ) {
			if ( ! is_array( $entry ) || '' === (string) ( $entry['path'] ?? '' ) ) {
				throw new RuntimeException( "File stage '{$stage_key}' manifest returned a file list with an entry that is not a file." );
			}
		}

		if ( [] === $data['files'] ) {
			return null;
		}

		$scoped = $this->scope_stage_manifest( $data['files'], $stage_key, $record );

		return [] === $scoped ? null : $scoped;
	}

	/**
	 * Count a failure against one file and decide whether it gets another try.
	 *
	 * Three strikes, each retried from byte 0, then the file is marked failed and the
	 * stage's failure count stops the run with a total. Failing the migration on the
	 * first bad file would throw away every byte already moved.
	 *
	 * @param array<string, mixed> $meta The file's progress entry.
	 *
	 * @return bool Always true: there is more work, if only the next file.
	 */
	/**
	 * A write the destination refuses by rule (a name or folder it never writes) will be refused again, so
	 * it is skipped and logged rather than struck three times and counted as a failed file.
	 */
	private static function is_refused_write( \Throwable $e ): bool {
		if ( $e instanceof HttpRequestException ) {
			return str_ends_with( $e->getMessage(), 'Remote said: ' . WriteFailure::REFUSED_MESSAGE );
		}

		return $e instanceof FileRefusedException;
	}

	private function skip_refused_file( int $migration_id, string $path, array $meta, string $stage, MigrationLogger $logger ): bool {
		$logger->warning( "File skipped, the destination never writes it: {$path}" );
		$this->update_file_progress( $migration_id, [ $path => array_merge( $meta, [ 'done' => true, 'skipped' => true ] ) ], 0, $stage, [ $path ] );

		return true;
	}

	private function strike_file( int $migration_id, string $path, array $meta, string $stage, \Throwable $error, MigrationLogger $logger ): bool {
		$error_count = (int) ( $meta['errors'] ?? 0 ) + 1;
		$logger->error( "File transfer failed (attempt {$error_count}/3): {$path}", [
			'error' => $error->getMessage(),
		] );

		if ( $error_count >= 3 ) {
			$logger->error( "Skipping file after 3 failed attempts: {$path}" );
			$entry = array_merge( $meta, [
				'done'   => true,
				'failed' => true,
				'errors' => $error_count,
			] );
		} else {
			$entry = array_merge( $meta, [
				'offset' => 0,
				'done'   => false,
				'errors' => $error_count,
			] );
		}

		$this->update_file_progress( $migration_id, [ $path => $entry ], 0, $stage, [ $path ] );

		return true;
	}

	/**
	 * Recover when the remote refuses a file this migration still has queued.
	 *
	 * The refusal is correct, since the queued path is not on the manifest the remote
	 * built. But the queue and manifest can disagree without operator action (a
	 * restored database, a source that changed mid-transfer), and failing the
	 * migration over one missing path is the wrong trade.
	 *
	 * So the manifest is re-fetched and reconciled. A vanished path loses its row and
	 * the worker moves on; a path still offered is retried against a matching state.
	 *
	 * @return bool True when the path survived the refresh and may be retried.
	 */
	private function heal_manifest_drift( int $migration_id, MigrationRecord $record, string $path, string $stage_key, HttpRequestException $refused, MigrationLogger $logger ): bool {
		// Two answers are drift: the scope refusal, and a file the source's cached
		// scope lists but that is gone from disk. A manifest refresh settles both. A 401
		// or 403 for any other reason is an auth or policy problem it cannot fix.
		if ( ! in_array( $refused->remote_code(), [ 'shiftora_file_not_allowed', 'shiftora_file_error' ], true ) ) {
			throw $refused;
		}

		$connection = $this->connections->find_by_id( $record->connection_id );
		if ( ! $connection ) {
			throw $refused;
		}

		$logger->warning(
			sprintf(
				"Source no longer offers '%s' for stage '%s', refreshing the manifest and dropping the stale entry.",
				$path,
				$stage_key
			)
		);

		try {
			$manifest = $this->fetch_pull_manifest( $connection, $record, $stage_key, $logger );
		} catch ( DeadlineExceeded $deadline ) {
			throw $deadline;
		} catch ( RuntimeException $unavailable ) {
			// The source cannot say what it has right now. Keep the original refusal as
			// the truthful error rather than continue against an unknown scope.
			$logger->error( sprintf( 'Could not refresh the manifest to resolve drift: %s', $unavailable->getMessage() ) );
			throw $refused;
		}

		// A source that suddenly offers nothing for a stage it was serving is more
		// likely broken than emptied. Dropping every row would complete the stage with
		// files still to send.
		if ( null === $manifest || [] === $manifest ) {
			$logger->error( sprintf( "The refreshed manifest for stage '%s' came back empty; keeping the original refusal.", $stage_key ) );
			throw $refused;
		}

		$state = $this->build_file_offsets_state( $manifest );
		$this->refresh_file_stage( $migration_id, $stage_key, $manifest, $state['bytes_total'], $state['offsets'] );

		// The manifest is a list of entries; membership is checked on the path-keyed
		// map built from it.
		if ( array_key_exists( $path, $state['offsets'] ) ) {
			$logger->warning( sprintf( "Source still offers '%s'; retrying the request against refreshed state.", $path ) );
			return true;
		}

		$logger->info( sprintf( "Stage '%s' now advertises %d file(s) after dropping the stale entry.", $stage_key, count( $manifest ) ) );

		return false;
	}

	/**
	 * Advance to the next enabled file stage after $after_stage completes or is
	 * skipped. With none left, the migration is marked Complete.
	 *
	 * @param string $after_stage The finished stage key. When empty, falls back to the current_stage recorded in the DB.
	 */
	private function advance_to_next_file_stage( int $id, MigrationLogger $logger, string $after_stage = '', bool $mark_current_complete = true ): void {
		$record  = $this->load( $id );
		$stages  = $record->stages();

		// Prefer the explicit hint; fall back to what is recorded in the DB.
		$current    = '' !== $after_stage ? $after_stage : ( $record->current_stage ?? '' );
		if ( $mark_current_complete && in_array( $current, FileStages::keys(), true ) ) {
			$this->complete_file_stage( $id, $current, $record, $logger );
			$record = $this->load( $id );
		}

		$next_stage = $this->next_enabled_file_stage( $stages, $current, $record );

		if ( null !== $next_stage ) {
			$this->prepare_file_stage_generic( $id, $record, $next_stage );
			return;
		}

		if ( $this->handoff_to_database_stage( $id, $record, $logger ) ) {
			return;
		}

		$this->complete_migration( $id );

		// Only after the commit has landed. complete_migration() can fail the
		// migration or lose a race with a cancel, and saying "finished" over either would
		// mislead on the status screen.
		$final = $this->load( $id );
		if ( MigrationStatus::Completed === $final->status ) {
			$logger->info( 'All selected stages complete. Migration finished.' );
		} else {
			$logger->info( sprintf(
				'All selected stages processed, but the migration resolved as "%s" rather than completing.',
				(string) $final->status
			) );
		}
	}

	/** Mark the migration Completed with a completed_at timestamp. */
	private function complete_migration( int $id ): void {
		$wpdb = $this->wpdb;

		$record = $this->load( $id );

		// The per-file table decides: pending or failed rows in any stage block completion.
		$blocking = $this->file_progress->blocking( $id );
		if ( $blocking['pending'] > 0 || $blocking['failed'] > 0 ) {
			$failed_count = $blocking['failed'];
			$message      = $failed_count > 0
				? sprintf( 'Migration stopped because %d file(s) failed during transfer.', $failed_count )
				: 'Migration cannot be marked completed while file transfer work is still pending.';
			$this->fail( $id, $message );
			return;
		}

		// Verify every planned stage was completed.
		$plan      = $record->planned_stage_order();
		$completed = $record->completed_stages();
		foreach ( $plan as $stage_key ) {
			if ( ! isset( $completed[ $stage_key ] ) ) {
				$this->fail( $id, "Migration cannot complete: stage '{$stage_key}' was never marked completed." );
				return;
			}
		}

		// What this run committed, as opposed to what it planned. A completed stage is
		// the worker's own belief, written before the commit boundary, so it is not proof
		// that the rename ran.
		$database_planned = ! empty( $record->tables() );
		$committed_tables = $this->committed_table_count( $record );

		if ( $database_planned && null === $committed_tables && ! $record->dry_run ) {
			$this->fail(
				$id,
				'Migration cannot complete: the database stage finished but no finalize receipt was recorded, so the tables were never committed.'
			);
			return;
		}

		$now           = current_time( 'mysql', true );
		[ $owner_checked, $owner_token ] = $this->ownership_guard();

		// The commit is a guarded compare-and-swap that only lands while the row is
		// still owned and running. A cancel or pause that raced the final chunk wins, and
		// no post-completion work runs over a migration the operator resolved.
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$committed = (bool) $wpdb->query( $wpdb->prepare(
			"UPDATE %i
			    SET status = %s, completed_at = %s, updated_at = %s
			  WHERE id = %d AND status = 'running' AND (%d = 0 OR worker_lock_owner = %s)",
			$wpdb->prefix . 'shiftora_migrations',
			MigrationStatus::Completed->value,
			$now,
			$now,
			$id,
			$owner_checked,
			$owner_token
		) );

		if ( ! $committed ) {
			$logger = new MigrationLogger( $id );
			$logger->warning(
				'Worker lost ownership at the completion step; leaving the migration as the operator resolved it.'
			);
			return;
		}

		// Post-completion deployment verification.
		$this->verify_deployment_consistency( $id, $record );

		// The run is over, so its lease-scoped state goes with it.
		$this->release_transfer_state( $id );

		// Completion logs too, so a healthy run leaves something to read back and attach
		// to a support question.
		$migrated_tables = ( null !== $committed_tables && $committed_tables > 0 )
			? array_slice( $record->tables(), 0, (int) $committed_tables )
			: [];

		// processed_rows is what the worker moved. total_rows is the plan's figure and
		// is 0 for every pull, which would print "0 row(s)" after thousands were imported.
		$rows_moved = (int) $record->processed_rows;

		$completion_log = ( null !== $committed_tables && $committed_tables > 0 )
			? sprintf(
				'%s migration completed: %d table(s) committed, %s, %d file(s).',
				mb_strtoupper( $record->direction->value ),
				count( $migrated_tables ),
				$rows_moved > 0
					? sprintf( '%d row(s) transferred', $rows_moved )
					: 'rows were not counted',
				(int) $record->total_files
			)
			: sprintf(
				'%s migration completed: no tables were committed, %d file(s) transferred.',
				mb_strtoupper( $record->direction->value ),
				(int) $record->total_files
			);

		// A pull imports what the source advertises and leaves the rest of this site
		// alone. "14 table(s)" read later would look like a complete replacement.
		$left_out = [];
		if ( MigrationDirection::Pull === $record->direction ) {
			// A pull carries the source's bare names while this site holds prefixed ones, so
			// diffing them as strings matches nothing and reports every table as left behind.
			// Compare bare names on both sides.
			$touched = array_map(
				function ( $table ): string {
					return $this->exporter->strip_prefix( (string) $table );
				},
				$migrated_tables
			);

			$left_out = array_values(
				array_filter(
					$this->exporter->get_wp_tables(),
					function ( string $table ) use ( $touched ): bool {
						return ! in_array( $this->exporter->strip_prefix( $table ), $touched, true );
					}
				)
			);
			if ( $left_out ) {
				$completion_log .= sprintf(
					' %d local table(s) were not part of the source set and were left untouched: %s.',
					count( $left_out ),
					implode( ', ', $left_out )
				);
			}
		}

		( new MigrationLogger( $id ) )->info( $completion_log );

		// Post-import stabilization: the swapped DB has to be reflected in the object
		// cache and rewrite rules, and the host's page cache has to drop the previous
		// site's rendered HTML.
		CacheFlusher::flush_silently();

		if ( function_exists( 'do_action' ) ) {
			/**
			 * Fires when a migration has been fully completed.
			 *
			 * @param int              $migration_id Migration ID.
			 * @param \Shiftora\Migration\MigrationRecord $record       The completed migration record.
			 */
			do_action( 'shiftora_migration_complete', $id, $record );
		}

		// Completion telemetry is emitted once, by verify_deployment_consistency(),
		// which knows whether the run was clean or degraded.
	}

	/**
	 * How many tables this run committed, or null when nothing proves a commit.
	 *
	 * Non-null means a durable receipt on this site proves this transfer swapped that
	 * many tables. Receipts are keyed by the transfer's uuid, so an earlier run cannot
	 * answer for this one.
	 *
	 * Zero is a real count (a rehearsal, or nothing selected). Null fails the run in
	 * complete_migration() and zero completes it; collapsing them would report a
	 * database still in staging as a migration that moved no rows.
	 *
	 */
	private function committed_table_count( MigrationRecord $record ): ?int {
		// A dry run never finalizes and a run with nothing selected has nothing to
		// finalize, so no receipt exists. Both committed nothing, which is 0 and not the
		// null that means a commit could not be proven.
		if ( $record->dry_run || empty( $record->tables() ) ) {
			return 0;
		}

		return $this->importer->committed_finalize_tables(
			$record->id,
			(string) ( $record->transfer_uuid ?? '' )
		);
	}

	/**
	 * Run non-blocking post-completion consistency checks. They run after
	 * status=Completed is set and log info or warning, never failure.
	 */
	private function verify_deployment_consistency( int $id, MigrationRecord $record ): void {
		$logger = new MigrationLogger( $id );
		$issues = [];

		// 1. Site URL and home URL consistency.
		$site_url = untrailingslashit( get_site_url() );
		$home     = untrailingslashit( get_home_url() );
		if ( $site_url !== $home ) {
			$logger->warning( 'Deployment verification: siteurl and home differ.', [
				'siteurl' => $site_url,
				'home'    => $home,
			] );
			$issues[] = 'site_url_mismatch';
		} else {
			$logger->info( 'Deployment verification: site URL consistent.', [
				'url' => $site_url,
			] );
		}

		// 2. Plugin file integrity.
		$shiftora_dir = defined( 'SHIFTORA_DIR' ) ? SHIFTORA_DIR : '';
		if ( '' !== $shiftora_dir && is_dir( $shiftora_dir ) ) {
			$critical = [ 'shiftora.php', 'src/Core/Plugin.php', 'src/Core/Installer.php' ];
			foreach ( $critical as $cf ) {
				$full = $shiftora_dir . '/' . $cf;
				if ( ! is_file( $full ) ) {
					$logger->error( 'Deployment verification: critical plugin file missing.', [
						'file' => $cf,
					] );
					$issues[] = 'plugin_file_missing_' . str_replace( '/', '_', $cf );
				}
			}
			// Each missing file is recorded under its own key, so check for the prefix.
			$missing_plugin_files = array_filter(
				$issues,
				static fn( string $issue ): bool => str_starts_with( $issue, 'plugin_file_missing_' )
			);
			if ( [] === $missing_plugin_files ) {
				$logger->info( 'Deployment verification: plugin directory integrity passed.' );
			}
		}

		// 3. Sample file existence check from completed file stages.
		$completed_stages = $record->completed_stages();
		foreach ( FileStages::keys() as $stage_key ) {
			if ( ! isset( $completed_stages[ $stage_key ] ) ) {
				continue;
			}
			$base_dir = FileStages::base_dir( $stage_key );
			if ( '' === $base_dir || ! is_dir( $base_dir ) ) {
				$logger->warning( 'Deployment verification: file stage base directory not found.', [
					'stage' => $stage_key,
					'dir'   => $base_dir,
				] );
				$issues[] = 'stage_dir_missing_' . $stage_key;
				continue;
			}
			$logger->info( 'Deployment verification: file stage base directory exists.', [
				'stage' => $stage_key,
				'dir'   => $base_dir,
			] );

			// A directory existing is not evidence that anything arrived in it; a
			// stage that promoted nothing can leave the site visibly empty. A stage the run
			// tracked files for must have accounted for every one. A stage that tracked none
			// resolved to an empty selection, and an empty folder is the honest outcome.
			$tally = $this->file_progress->summarize( $id, $stage_key );
			$total = (int) ( $tally['total_files'] ?? 0 );

			if ( 0 === $total ) {
				$logger->info( 'Deployment verification: file stage carried no files.', [
					'stage' => $stage_key,
				] );
				continue;
			}

			$transferred = (int) ( $tally['transferred_files'] ?? 0 );

			if ( $transferred < $total ) {
				$logger->warning( 'Deployment verification: file stage did not account for every file it staged.', [
					'stage'       => $stage_key,
					'staged'      => $total,
					'transferred' => $transferred,
				] );
				$issues[] = 'stage_files_missing_' . $stage_key;
				continue;
			}

			// A promoted file has a digest it was verified against, so a sample can be
			// re-checked for real rather than trusted.
			$sample     = $this->sample_transferred_files( $id, $stage_key, $base_dir, $logger );

			if ( [] !== $sample ) {
				$logger->warning( 'Deployment verification: transferred files are not on disk as recorded.', [
					'stage' => $stage_key,
					'count' => count( $sample ),
				] );
				$issues[] = 'stage_files_not_on_disk_' . $stage_key;
			} else {
				$logger->info( 'Deployment verification: transferred files are present on disk.', [
					'stage' => $stage_key,
				] );
			}
		}

		if ( ! empty( $issues ) ) {
			$logger->warning( 'Deployment verification completed with issues.', [
				'issues' => $issues,
			] );
		} else {
			$logger->info( 'Deployment verification passed all consistency checks.' );
		}

		MigrationObservability::emit_migration_completed(
			(string) $id,
			empty( $issues ) ? 'success' : 'degraded',
			count( $record->tables() ),
			(int) $record->total_files,
			(int) $record->bytes_transferred,
			empty( $issues ) ? 0.0 : 1.0
		);
	}

	/**
	 * Re-check a bounded sample of transferred files against the filesystem.
	 *
	 * The progress table says a file was promoted; this asks the disk. It compares a
	 * digest, since a partial write also leaves a file in place. The sample is the
	 * first files in stage order, where a stage that broke immediately shows. A pull
	 * of a stage the destination already holds acks most entries without rewriting
	 * them, so matching files prove little, while a missing file is reported.
	 *
	 * @return string[] Paths that did not hold what the run recorded.
	 */
	private function sample_transferred_files( int $id, string $stage_key, string $stage_base, MigrationLogger $logger ): array {
		$importer = FileStages::writer( $stage_key, $stage_base );
		$missing  = [];

		foreach ( $this->file_progress->sample_done( $id, $stage_key ) as $row ) {
			$path = (string) ( $row['virtual_path'] ?? '' );
			$hash = (string) ( $row['content_hash'] ?? '' );

			if ( '' === $path || '' === $hash ) {
				continue;
			}

			if ( ! $importer->already_transferred( $path, (int) $row['byte_total'], $hash ) ) {
				$missing[] = $path;
			}
		}

		if ( [] !== $missing ) {
			$logger->warning( 'Deployment verification: a transferred file is not on disk as recorded.', [
				'stage'   => $stage_key,
				'missing' => array_slice( $missing, 0, 5 ),
				'count'   => count( $missing ),
			] );
		}

		return $missing;
	}

	/**
	 * Process the next file chunk for a file-stage migration. Called by the
	 * `shiftora_process_file_chunk` Action Scheduler action.
	 *
	 * Each chunk's SHA-256 is verified by the stage's writer before any
	 * write, and the full file's by verify_file_hash() after the last. A mismatch
	 * throws RuntimeException, which MigrationJob marks failed.
	 *
	 * @return bool True if more chunks remain, false when done.
	 *
	 * @throws RuntimeException On connection error or integrity failure.
	 */
	public function process_next_file_chunk( int $migration_id, ?float $deadline = null ): bool {
		if ( ! $this->acquire_migration_lock( $migration_id ) ) {
			return true;
		}

		$this->active_deadline = $deadline;

		try {
			$this->guard_deadline();

			$logger = new MigrationLogger( $migration_id );
			$record = $this->load( $migration_id );

		if ( ! in_array( $record->status, [ MigrationStatus::Running, MigrationStatus::Pending ], true ) ) {
			return false;
		}

		// Per-file progress lives in the per-file table. A legacy record still carrying
		// the JSON blob is converted to rows once, on first load, so the worker reads
		// bounded rows and never the whole manifest.
		$raw_offsets = (string) ( $record->file_offsets_json ?? '' );
		$stage_label = (string) ( $record->current_stage ?? '' );
		$has_rows    = $this->file_progress->has_rows( $migration_id, $stage_label );

		if ( ! $has_rows && '' !== $raw_offsets && 'database' !== $stage_label ) {
			$decoded = json_decode( $raw_offsets, true );
			if ( JSON_ERROR_NONE !== json_last_error() ) {
				$logger->critical( 'Corrupt file_offsets_json detected.', [
					'migration_id' => $migration_id,
					'json_error'   => json_last_error_msg(),
				] );
				$this->fail( $migration_id, 'Migration file offset state is corrupt.' );
				return false;
			}
			if ( is_array( $decoded ) && [] !== $decoded ) {
				$this->file_progress->ensure_seeded( $migration_id, $stage_label, $decoded );
				$has_rows = true;
			}
		}

		if ( ! $has_rows ) {
			if ( 'database' === $stage_label ) {
				// The database stage is active, which is not a file worker concern.
				return false;
			}

			if ( '[]' === $raw_offsets ) {
				// Legitimately zero files: log and advance.
				$logger->info( "File stage '{$stage_label}': zero files to process, advancing." );
				$this->advance_to_next_file_stage( $migration_id, $logger );
			} else {
				$logger->error( "File stage '{$stage_label}': file_offsets_json missing unexpectedly." );
				$this->fail( $migration_id, "File stage '{$stage_label}' offset state is missing." );
			}
			$next_record = $this->load( $migration_id );
			return in_array( $next_record->status, [ MigrationStatus::Running, MigrationStatus::Pending ], true )
				&& ! empty( $next_record->file_offsets_json );
		}

		// Bulk deduplication: scan pending files in bounded batches instead of loading
		// the whole offsets map each chunk, which turns N read/stat/write cycles on a
		// large uploads pull into a few indexed lookups. Matches are marked done, so
		// bandwidth goes only to what is missing.
		if ( 'uploads' === $stage_label ) {
			$deduped         = 0;
			$deduped_entries = [];

			$stage_base = FileStages::base_dir( $stage_label );
			$importer   = FileStages::writer( $stage_label, $stage_base );

			$batch = $this->file_progress->next_pending_batch( $migration_id, $stage_label );
			foreach ( $batch as $row ) {
				$path = (string) ( $row['virtual_path'] ?? '' );
				if ( '' === $path ) {
					continue;
				}
				$meta = $this->file_progress->row_to_meta( $row );
				if ( $importer->already_transferred( $path, (int) $meta['size'], (string) $meta['hash'] ) ) {
					$meta['done']             = true;
					$deduped_entries[ $path ] = $meta;
					$deduped++;
				}
			}

			if ( $deduped > 0 ) {
				$logger->info( sprintf(
					'Bulk deduplication: %d pending file(s) already on disk, skipped.',
					$deduped
				) );
				$this->update_file_progress(
					$migration_id,
					$deduped_entries,
					0,
					$stage_label,
					array_keys( $deduped_entries )
				);
			}
		}

		// Find the first incomplete file: the lowest-sequence pending row. The row
		// carries every field the transfer needs, so the worker holds one file's
		// metadata and not the whole manifest.
		$current_path = null;
		$current_meta = null;
		$next_file    = $this->file_progress->next_pending( $migration_id, $stage_label );
		if ( null !== $next_file ) {
			$current_path = (string) $next_file->virtual_path;
			$current_meta = $this->file_progress->row_to_meta( $next_file );
		}

		if ( null === $current_path ) {
			$blocking = $this->file_progress->stage_blocking( $migration_id, $stage_label );

			if ( $blocking['failed'] > 0 ) {
				$this->fail(
					$migration_id,
					sprintf( 'Migration stopped because %d file(s) failed during the %s stage.', $blocking['failed'], $stage_label )
				);
				return false;
			}

			$this->advance_to_next_file_stage( $migration_id, $logger );
			$next_record = $this->load( $migration_id );
			return in_array( $next_record->status, [ MigrationStatus::Running, MigrationStatus::Pending ], true )
				&& ! empty( $next_record->file_offsets_json );
		}

		// Resolve the base directory for the current file migration stage.
		$current_stage = $record->current_stage ?? 'uploads';
		$stage_base    = FileStages::base_dir( $current_stage );

		// Deduplication: skip if already intact on disk, uploads stage only. Code
		// stages are always re-imported fresh without pre-wiping, so it is unsafe there.
		$importer = FileStages::writer( $current_stage, $stage_base );

		if ( 'uploads' === $current_stage &&
			$importer->already_transferred(
				$current_path,
				(int) $current_meta['size'],
				(string) $current_meta['hash']
			)
		) {
			$logger->info( "File already transferred (deduplicated): {$current_path}" );
			$current_meta['done'] = true;
			$this->update_file_progress( $migration_id, [ $current_path => $current_meta ], 0, $current_stage, [ $current_path ] );
			return true;
		}

		// Resolve connection.
		$connection = $this->connections->find_by_id( $record->connection_id );
		if ( ! $connection ) {
			throw new RuntimeException( "Connection #{$record->connection_id} not found." );
		}

		$offset = (int) ( $current_meta['offset'] ?? 0 );
		$operation_started_ns = hrtime( true );

		// Pull: fetch a chunk from the remote /export/media-stream and write it locally.

		// Batch-collect: gather small pending files up to the budget and fetch them in
		// one POST to /export/file-batch. Files marked skip_batch (the remote could not
		// serve them in a batch) fall through to the single-file stream path.
		if ( 0 === $offset &&
			(int) $current_meta['size'] <= self::BATCH_FILE_THRESHOLD &&
			! ( $current_meta['skip_batch'] ?? false )
		) {
			$batch_budget = $this->batch_budget_bytes( $migration_id, $current_stage );

			$batch_paths   = [ $current_path ];
			$batch_entries = [ $current_path => $current_meta ];
			$budget_used   = (int) $current_meta['size'];

			foreach ( $this->file_progress->next_pending_batch( $migration_id, $current_stage ) as $candidate ) {
				$fpath = (string) ( $candidate['virtual_path'] ?? '' );
				if ( '' === $fpath || $fpath === $current_path ) {
					continue;
				}
				$fmeta = $this->file_progress->row_to_meta( $candidate );
				if ( (int) $fmeta['offset'] > 0 ) {
					break;
				}
				// Skip files the remote flagged as too large for batching.
				if ( $fmeta['skip_batch'] ) {
					continue;
				}
				if ( (int) $fmeta['size'] > self::BATCH_FILE_THRESHOLD ) {
					continue;
				}
				if ( count( $batch_paths ) >= self::BATCH_MAX_FILES ) {
					break;
				}
				if ( $budget_used + (int) $fmeta['size'] > $batch_budget ) {
					break;
				}
				// Inline dedup, uploads stage only: skip files already intact on disk.
				if ( 'uploads' === $current_stage && $importer->already_transferred( $fpath, (int) $fmeta['size'], (string) $fmeta['hash'] ) ) {
					$logger->info( "File already on disk (batch deduplication): {$fpath}" );
					$fmeta['done'] = true;
					$this->update_file_progress( $migration_id, [ $fpath => $fmeta ], 0, $current_stage, [ $fpath ] );
					continue;
				}
				$batch_paths[]            = $fpath;
				$batch_entries[ $fpath ]  = $fmeta;
				$budget_used             += (int) $fmeta['size'];
			}

			if ( count( $batch_paths ) >= 2 ) {
				return $this->process_file_batch_pull(
					$migration_id,
					$record,
					$batch_paths,
					$batch_entries,
					$importer,
					$connection,
					$current_stage,
					$logger
				);
			}
		}

		$route_path = '/shiftora/v1/export/media-stream';
		$query      = http_build_query( [
			'stage'      => $current_stage,
			'file'       => $current_path,
			'offset'     => $offset,
			'chunk_size' => FileExporter::DEFAULT_CHUNK_SIZE,
		] );
		$full_url  = rtrim( $connection->remote_url, '/' ) . '/wp-json' . $route_path . '?' . $query;

		$this->guard_deadline();
		try {
			$response = $this->fetch_with_retry(
				function () use ( $connection, $route_path, $query, $full_url ) {
					$headers = $this->request_authenticator->create_auth_headers_for_connection( $connection, 'GET', $route_path . '?' . $query );

					return $this->http->get( $full_url, [
						...$headers,
					], $this->request_timeout() );
				},
				3,
				$logger,
				"media stream"
			);
		} catch ( HttpRequestException $refused ) {
			$healed = $this->heal_manifest_drift( $migration_id, $record, $current_path, $current_stage, $refused, $logger );

			if ( ! $healed ) {
				// The source no longer offers the file and its row is gone, so the next chunk
				// picks a different one.
				return true;
			}
			// The file is still on offer, so the refusal was about scope and not the file.
			// Re-issue the request against the refreshed state.
			$response = $this->fetch_with_retry(
				function () use ( $connection, $route_path, $query, $full_url ) {
					$headers = $this->request_authenticator->create_auth_headers_for_connection( $connection, 'GET', $route_path . '?' . $query );

					return $this->http->get( $full_url, [
						...$headers,
					], $this->request_timeout() );
				},
				3,
				$logger,
				"media stream"
			);
		}

		$chunk = json_decode( $response['body'], true );

		if ( ! is_array( $chunk ) || ! isset( $chunk['data'], $chunk['chunk_hash'] ) ) {
			throw new RuntimeException(
				"Invalid file chunk response from remote for: {$current_path}"
			);
		}

		// Per-chunk integrity check and write, then the full-file check on the last
		// chunk. A failure (hash mismatch, bad payload, write error) is the file's
		// problem, not the migration's: it is retried from byte 0 up to three times, then
		// marked failed, and the stage's failure count stops the run. The write returns
		// the accepted length, so the pull cursor advances past it even on a replay.
		try {
			$accepted = $importer->write_chunk(
				$current_path,
				(string) $chunk['data'],
				$offset,
				(string) $chunk['chunk_hash'],
				! empty( $chunk['compressed'] )
			);

			$new_offset = $offset + $accepted;
			$has_more   = ! empty( $chunk['has_more'] );

			if ( ! $has_more ) {
				// Full-file integrity check (throws and deletes the corrupt file on mismatch).
				$importer->verify_file_hash( $current_path, (string) $current_meta['hash'] );
			}
		} catch ( \RuntimeException $file_ex ) {
			if ( self::is_refused_write( $file_ex ) ) {
				return $this->skip_refused_file( $migration_id, $current_path, $current_meta, $current_stage, $logger );
			}
			return $this->strike_file( $migration_id, $current_path, $current_meta, $current_stage, $file_ex, $logger );
		}

		$operation_elapsed_ms = (int) round( ( hrtime( true ) - $operation_started_ns ) / 1_000_000 );
		$this->record_file_operation(
			$migration_id,
			'import',
			$current_path,
			$accepted,
			$operation_elapsed_ms
		);

		if ( ! $has_more ) {
			$current_meta['offset'] = $new_offset;
			$current_meta['done']   = true;
			$logger->info( "File transferred and verified: {$current_path}" );
		} else {
			$current_meta['offset'] = $new_offset;
		}

		$this->update_file_progress( $migration_id, [ $current_path => $current_meta ], $accepted, $current_stage, [ $current_path ] );

		// All-done and failed checks roll up through SQL aggregates, one read instead
		// of a manifest scan.
		$blocking = $this->file_progress->stage_blocking( $migration_id, $current_stage );

		if ( $blocking['failed'] > 0 ) {
			$this->fail(
				$migration_id,
				sprintf( 'Migration stopped because %d file(s) failed during the %s stage.', $blocking['failed'], $current_stage )
			);
			return false;
		}

		if ( 0 === $blocking['pending'] ) {
			$this->advance_to_next_file_stage( $migration_id, $logger );
			$next_record = $this->load( $migration_id );
			return in_array( $next_record->status, [ MigrationStatus::Running, MigrationStatus::Pending ], true )
				&& ! empty( $next_record->file_offsets_json );
		}

		return true;
		} finally {
			$this->active_deadline = null;
			$this->release_migration_lock( $migration_id );
		}
	}

	/**
	 * Complete one file stage. Every completion path (no pending files, a finishing
	 * chunk, both batch paths, a resume, the skip paths) goes through here so they
	 * reach the same decision.
	 *
	 * It only marks the stage complete. Destination files the source manifest does
	 * not list are never removed: a pull onto a site with other content must not
	 * delete that site's own files.
	 */
	private function complete_file_stage( int $id, string $stage_key, MigrationRecord $record, MigrationLogger $logger ): void {
		// Clean up abandoned scratch before the stage is declared done, so a crash
		// between the two leaves a stage that is still resumable.
		$this->cleanup_completed_file_stage( $id, $record, $stage_key, $logger );

		$this->mark_stage_completed( $id, $stage_key );
	}

	/**
	 * Clean up the scratch files this transfer owns: partial writes a worker
	 * abandoned and temporaries the exporter staged. Excluded files, the engine's
	 * runtime directories and everything the destination brought are never touched.
	 */
	private function cleanup_completed_file_stage( int $id, MigrationRecord $record, string $stage_key, MigrationLogger $logger ): void {
		if ( $record->dry_run ) {
			return;
		}

		// A stage the source advertised nothing for transferred no files. The skip
		// paths reach this call too, and an empty uploads stage would otherwise cost a
		// full recursive traversal of the base.
		if ( ! $this->file_progress->has_rows( $id, $stage_key ) ) {
			return;
		}

		$stage_base = FileStages::base_dir( $stage_key );
		if ( '' === $stage_base ) {
			return;
		}

		$tally = $this->file_progress->summarize( $id, $stage_key );
		if ( (int) ( $tally['skipped_files'] ?? 0 ) > 0 ) {
			$logger->warning(
				sprintf(
					'Stage "%s": %d file(s) moved and %d skipped because the destination never writes them. Skipped files are not counted as moved.',
					$stage_key,
					(int) $tally['transferred_files'],
					(int) $tally['skipped_files']
				)
			);
		}

		$result = CompleteFileStageOutcome::cleanup_stage(
			$stage_base,
			static function ( string $path ) use ( $logger, $stage_key ): bool {
				if ( ! FileStages::is_protected( $stage_key, $path ) ) {
					return false;
				}

				$logger->debug(
					sprintf(
						'Stage "%s" cleanup skipped %s: it is part of the running engine.',
						$stage_key,
						$path
					)
				);

				return true;
			},
			// An age floor, because this sweeps a directory another transfer may be writing
			// into right now. A write in progress has a current mtime, so the floor keeps a
			// pull finishing a stage from removing another run's scratch.
			self::ABANDONED_SCRATCH_MIN_AGE_SECONDS
		);

		if ( $result['removed'] > 0 ) {
			$logger->info(
				sprintf(
					'Stage "%s" cleanup: removed %d abandoned transfer file(s).',
					$stage_key,
					$result['removed']
				)
			);
		}

		if ( $result['incomplete'] ) {
			// Logged because this is the one path where litter is left behind silently, and
			// stray .shiftora-tmp files in an uploads tree are otherwise hard to explain.
			$logger->debug(
				sprintf(
					'Stage "%s" cleanup did not finish: part of %s could not be read. The transfer is unaffected.',
					$stage_key,
					$stage_base
				)
			);
		}
	}

	/**
	 * Persist file transfer progress to the per-file table and the migration row.
	 *
	 * Changed paths are upserted as single rows and the counters come from a SQL
	 * aggregate over the stage. file_offsets_json is only written at stage boundaries.
	 *
	 * @param array<string, array<mixed>> $changed_entries Sparse map of the entries mutated this chunk (path => meta, each carrying its manifest `sequence`).
	 * @param int                         $bytes_written   Raw bytes just written, for the progress events.
	 * @param string[]                    $changed_paths   Paths mutated since the last write.
	 */
	private function update_file_progress( int $id, array $changed_entries, int $bytes_written, string $stage_label, array $changed_paths = [] ): void {
		$wpdb = $this->wpdb;

		if ( [] === $changed_paths ) {
			$changed_paths = array_keys( $changed_entries );
		}
		$this->file_progress->upsert_changed( $id, $stage_label, $changed_entries, $changed_paths );

		$summary           = $this->file_progress->summarize( $id, $stage_label );
		$transferred_files = $summary['transferred_files'];
		$total_files       = $summary['total_files'];
		$bytes_total       = $summary['bytes_total'];
		$bytes_transferred = $summary['bytes_transferred'];
		$file_progress     = $summary['progress'];
		$now               = current_time( 'mysql', true );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$wpdb->query( $wpdb->prepare(
			'UPDATE %i
			    SET total_files        = %d,
			        transferred_files  = %d,
			        bytes_total        = %d,
			        bytes_transferred  = %d,
			        current_stage      = %s,
			        stage_progress     = %d,
			        last_heartbeat_at  = %s,
			        updated_at         = %s
			  WHERE id = %d',
			$wpdb->prefix . 'shiftora_migrations',
			$total_files,
			$transferred_files,
			$bytes_total,
			$bytes_transferred,
			$stage_label,
			$file_progress,
			$now,
			$now,
			$id
		) );

		MigrationObservability::emit_migration_progress(
			(string) $id,
			$stage_label,
			$transferred_files,
			$total_files,
			$bytes_written,
			0
		);
	}

	/**
	 * Record a file operation and emit performance warnings when thresholds are exceeded.
	 */
	private function record_file_operation(
		int $migration_id,
		string $type,
		string $path,
		int $size_bytes,
		int $duration_ms
	): void {
		$this->file_auditor->record( $type, $path, $size_bytes, $duration_ms );

		if ( $duration_ms > 5000 ) {
			MigrationObservability::emit_performance_warning(
				(string) $migration_id,
				'slow_operation',
				'file_' . $type . '_duration_ms',
				$duration_ms,
				5000
			);
		}

		if ( $size_bytes > 104_857_600 ) {
			MigrationObservability::emit_performance_warning(
				(string) $migration_id,
				'large_file',
				'file_' . $type . '_size_bytes',
				$size_bytes,
				104_857_600
			);
		}
	}

	/** Enqueue the next file-chunk job, falling back when Action Scheduler is unavailable or throws. */
	private function enqueue_file_chunk_job( int $id, MigrationLogger $logger ): bool {
		try {
			return MigrationJob::schedule_file( $id );
		} catch ( \Throwable $e ) {
			$logger->error( 'Action Scheduler file-chunk enqueue failed.', [
				'migration_id' => $id,
				'error'        => $e->getMessage(),
			] );
		}

		return false;
	}

	/**
	 * Pull a batch of small files from the remote source in one HTTP request.
	 *
	 * POSTs the relative paths to /export/file-batch and writes each returned payload
	 * with the same per-chunk and full-file SHA-256 checks as the single-file path.
	 * Files the remote marks `skipped_large` are tagged skip_batch so the next call
	 * takes them via the single-file stream endpoint.
	 *
	 * @param array<string, array<mixed>> $batch_entries Sparse map of the batch's metadata (each entry carries its manifest `sequence`).
	 * @param string[]                    $batch_paths   Relative paths to fetch.
	 *
	 * @return bool True if more work remains, false when the stage is complete.
	 *
	 * @throws RuntimeException On connection error or unrecoverable batch failure.
	 */
	private function process_file_batch_pull(
		int $migration_id,
		object $record,
		array $batch_paths,
		array $batch_entries,
		FileWriter $importer,
		object $connection,
		string $current_stage,
		MigrationLogger $logger
	): bool {
		$route_path = '/shiftora/v1/export/file-batch';
		$full_url   = rtrim( $connection->remote_url, '/' ) . '/wp-json' . $route_path;

		// One request for a set of paths, split out so the halving path below can call
		// it twice with less to ask for.
		$fetch_paths = function ( array $paths ) use ( $connection, $route_path, $full_url, $record, $current_stage, $logger ): array {
			$payload = wp_json_encode( [
				'stage' => $current_stage,
				'paths' => $paths,
			] );

			$response = $this->fetch_with_retry(
				function () use ( $connection, $route_path, $full_url, $payload ) {
					$headers = $this->request_authenticator->create_auth_headers_for_connection( $connection, 'POST', $route_path, (string) $payload );

					return $this->http->post( $full_url, (string) $payload, [
						'Content-Type' => 'application/json',
						...$headers,
					], $this->request_timeout() );
				},
				3,
				$logger,
				'file batch'
			);

			$data = json_decode( (string) ( $response['body'] ?? '' ), true );

			if ( ! is_array( $data ) || ! isset( $data['files'] ) || ! is_array( $data['files'] ) ) {
				throw new RuntimeException( 'Invalid response from remote /export/file-batch.' );
			}

			return $data['files'];
		};

		$sizer  = $this->transfer_sizing->load( $migration_id, $current_stage, $this->batch_encoded_ceiling() );
		$started = microtime( true );

		// A pull batch that times out should not end the migration. Retrying the same
		// request just repeats the timeout, and asking for fewer files solves it, so the
		// batch is halved. The sizer records that this batch was
		// too big so the next scheduled run assembles a smaller one.
		try {
			$this->log_batch_request( $logger, 'pull', $current_stage, $this->batch_raw_bytes_for_paths( $batch_paths, $batch_entries ), count( $batch_paths ), 'sent' );
			$files = $fetch_paths( $batch_paths );
		} catch ( HttpRequestException $e ) {
			if ( ! FileBatchSizer::should_split( count( $batch_paths ), $e->http_status() ) ) {
				$this->log_batch_request( $logger, 'pull', $current_stage, $this->batch_raw_bytes_for_paths( $batch_paths, $batch_entries ), count( $batch_paths ), 'refused', $e->http_status() );
				$this->record_batch_failure( $migration_id, $current_stage, $sizer, $e->http_status() );
				throw $e;
			}

			[ $head, $tail ] = FileBatchSizer::halve( $batch_paths );

			$this->log_batch_request( $logger, 'pull', $current_stage, $this->batch_raw_bytes_for_paths( $batch_paths, $batch_entries ), count( $batch_paths ), 'split', $e->http_status() );

			$logger->warning(
				sprintf(
					'Batch PULL did not complete (HTTP %d); retrying as %d + %d files.',
					$e->http_status(),
					count( $head ),
					count( $tail )
				),
				[
					'migration_id' => $migration_id,
					'stage'        => $current_stage,
					'files'        => count( $batch_paths ),
					'outcome'      => 'split',
				]
			);

			$files = array_merge( $fetch_paths( $head ), $fetch_paths( $tail ) );

			// Halved and it worked, so size was the problem. The halves are credited with
			// what they moved and how long it took, giving the time clamp a real rate. The
			// original attempt contributes nothing, having only shown this link could not
			// hold that many files at once.
			$asked_bytes = 0;
			foreach ( $batch_paths as $asked_path ) {
				$asked_bytes += (int) ( $batch_entries[ $asked_path ]['size'] ?? 0 );
			}

			$this->record_batch_sizing(
				$migration_id,
				$current_stage,
				$sizer->with_failure()->with_success( $asked_bytes, max( 0.001, microtime( true ) - $started ), $this->request_timeout() )
			);
		} catch ( DeadlineExceeded $e ) {
			// The worker's window closed, which says nothing about the link. Halving here
			// would drive a healthy pull to the floor one window at a time.
			throw $e;
		} catch ( \RuntimeException $e ) {
			$this->record_batch_sizing( $migration_id, $current_stage, $sizer->with_failure() );
			throw $e;
		}

		$bytes_total    = 0;
		$returned_paths = [];

		foreach ( $files as $file_entry ) {
			$path       = (string) ( $file_entry['path'] ?? '' );
			$raw_data   = (string) ( $file_entry['data'] ?? '' );
			$chunk_hash = (string) ( $file_entry['chunk_hash'] ?? '' );
			$compressed = ! empty( $file_entry['compressed'] );
			$file_start = hrtime( true );

			if ( '' === $path || '' === $chunk_hash ) {
				continue;
			}

			$returned_paths[] = $path;
			$manifest_hash    = (string) ( $batch_entries[ $path ]['hash'] ?? '' );

			try {
				$bytes_written = $importer->write_chunk( $path, $raw_data, 0, $chunk_hash, $compressed );
				$importer->verify_file_hash( $path, $manifest_hash );
				$elapsed_ms    = (int) round( ( hrtime( true ) - $file_start ) / 1_000_000 );

				$bytes_total                       += $bytes_written;
				$batch_entries[ $path ]['done']     = true;
				$batch_entries[ $path ]['offset']   = $bytes_written;
				$this->record_file_operation( $migration_id, 'import', $path, $bytes_written, $elapsed_ms );
				$logger->info( "Batch PULL: received and verified: {$path}" );
			} catch ( \RuntimeException $e ) {
				if ( self::is_refused_write( $e ) ) {
					$logger->warning( "Batch PULL: file skipped, the destination never writes it: {$path}" );
					$batch_entries[ $path ]['done']    = true;
					$batch_entries[ $path ]['skipped'] = true;
					continue;
				}

				$error_count = (int) ( $batch_entries[ $path ]['errors'] ?? 0 ) + 1;
				$logger->error( "Batch PULL: file error (attempt {$error_count}/3) {$path}: {$e->getMessage()}" );

				if ( $error_count >= 3 ) {
					$logger->error( "Batch PULL: skipping file after 3 failed attempts: {$path}" );
					$batch_entries[ $path ]['done']   = true;
					$batch_entries[ $path ]['failed'] = true;
					$batch_entries[ $path ]['errors'] = $error_count;
				} else {
					$batch_entries[ $path ]['offset'] = 0;
					$batch_entries[ $path ]['done']   = false;
					$batch_entries[ $path ]['errors'] = $error_count;
				}
			}
		}

		// Tag any file the remote could not serve (skipped_large or absent) with
		// skip_batch=true so batch-collect excludes it next time and it falls through to
		// the single-file stream path.
		$skipped_in_batch = array_diff( $batch_paths, $returned_paths );
		foreach ( $skipped_in_batch as $spath ) {
			if ( isset( $batch_entries[ $spath ] ) && ! ( $batch_entries[ $spath ]['done'] ?? false ) ) {
				$batch_entries[ $spath ]['skip_batch'] = true;
			}
		}

		$logger->info( sprintf(
			'Batch PULL: %d/%d files processed (%d bytes), %d queued for single-file fallback.',
			count( $returned_paths ),
			count( $batch_paths ),
			$bytes_total,
			count( $skipped_in_batch )
		) );

		// Recorded on the way out too: a pull that only succeeds would never write the
		// sizer, and the next migration would start at the default.
		//
		// Marked completed only once the stage is clear. One good batch does not prove the
		// stage was sent, and a partial run must not set the starting size for future ones.
		$still_blocking = $this->file_progress->stage_blocking( $migration_id, $current_stage );
		$this->record_batch_sizing(
			$migration_id,
			$current_stage,
			$sizer->with_success( $bytes_total, max( 0.001, microtime( true ) - $started ), $this->request_timeout() ),
			0 === $still_blocking['pending'] && 0 === $still_blocking['failed']
		);

		$this->update_file_progress( $migration_id, $batch_entries, $bytes_total, $current_stage, array_merge( $returned_paths, $skipped_in_batch ) );

		$blocking = $this->file_progress->stage_blocking( $migration_id, $current_stage );
		if ( $blocking['failed'] > 0 ) {
			$this->fail(
				$migration_id,
				sprintf( 'Migration stopped because %d file(s) failed during the %s stage.', $blocking['failed'], $current_stage )
			);
			return false;
		}

		if ( 0 === $blocking['pending'] ) {
			$this->advance_to_next_file_stage( $migration_id, $logger );
			$next_record = $this->load( $migration_id );
			return in_array( $next_record->status, [ MigrationStatus::Running, MigrationStatus::Pending ], true )
				&& ! empty( $next_record->file_offsets_json );
		}

		return true;
	}

	private static function server_error_streak_key( int $migration_id ): string {
		return 'shiftora_pull_5xx_' . $migration_id;
	}

	/**
	 * A source that keeps answering 5xx is usually choking on the size of the request,
	 * so the next window asks for a single table and half the rows per failure so far.
	 *
	 * @param array<int, array<string, mixed>> $cursors
	 *
	 * @return array{0: array<int, array<string, mixed>>, 1: int} The cursors and chunk size to send.
	 */
	private function shrink_after_server_errors( int $migration_id, array $cursors, int $chunk_size ): array {
		$level = (int) get_transient( self::server_error_shrink_key( $migration_id ) );
		if ( $level < 1 ) {
			return [ $cursors, $chunk_size ];
		}

		return [
			array_slice( $cursors, 0, 1 ),
			max( min( $chunk_size, self::PULL_SHRINK_FLOOR_ROWS ), $chunk_size >> min( $level, 16 ) ),
		];
	}

	/**
	 * Counts a pull window that ended on a 5xx. Attempts do not carry over between
	 * windows, so without this a persistent 5xx would retry for ever.
	 *
	 * The streak is what fails the run and an imported chunk clears it. The shrink
	 * level stays for the rest of the migration: a request shape the source already
	 * choked on is not worth sending again after every table that does go through.
	 *
	 * @throws HttpRequestException The source's error, once the limit is reached.
	 */
	private function count_server_error_window( int $migration_id, HttpRequestException $error, MigrationLogger $logger ): void {
		$shrink_key = self::server_error_shrink_key( $migration_id );
		set_transient( $shrink_key, (int) get_transient( $shrink_key ) + 1, DAY_IN_SECONDS );

		$key    = self::server_error_streak_key( $migration_id );
		$streak = (int) get_transient( $key ) + 1;

		if ( $streak >= self::PULL_SERVER_ERROR_LIMIT ) {
			delete_transient( $key );
			delete_transient( $shrink_key );
			$logger->error( sprintf( 'The other site answered with a server error in %d windows in a row; stopping.', $streak ) );
			throw $error;
		}

		set_transient( $key, $streak, DAY_IN_SECONDS );
	}

	private static function server_error_shrink_key( int $migration_id ): string {
		return 'shiftora_pull_shrink_' . $migration_id;
	}

	/**
	 * Fetch a resource with exponential-backoff retry (2^attempt seconds).
	 *
	 * An HttpRequestException carries the status, so a 403 or 404 (an answer, not an
	 * outage) propagates on the first attempt. Timeouts, 408, 429, 5xx and statusless
	 * transport errors are retried, as is a bare RuntimeException from callers outside
	 * the HTTP path. DeadlineExceeded is rethrown at once without sleeping.
	 *
	 * @param string $label What is being retried, named by the caller.
	 *
	 * @return mixed Return value of $fn on success.
	 *
	 * @throws RuntimeException When all attempts fail, or on the first attempt when the failure is not worth retrying.
	 */
	private function fetch_with_retry( callable $fn, int $max_attempts, MigrationLogger $logger, string $label = 'chunk' ): mixed {
		$last_exception = null;
		for ( $attempt = 1; $attempt <= $max_attempts; $attempt++ ) {
			try {
				return $fn();
			} catch ( DeadlineExceeded | DestinationOutOfSpace $e ) {
				// The worker budget is gone, and sleeping to retry would pass the scheduled
				// window. Yield so the next action picks up where the chunk stopped. A full
				// destination is rethrown as is, for the same reason as below.
				throw $e;
			} catch ( HttpRequestException $e ) {
				$this->fail_fast_when_out_of_space( $e, $logger, $label );

				$attempt_log = $this->transport_attempt_log( $attempt, $max_attempts, $e );

				// The transport knows whether this could ever succeed. A remote that refused
				// the request (wrong token, file outside the manifest, body rejected) will refuse
				// it identically each time, so retrying only adds backoff and log lines that bury
				// the reason. Surface it now with the status and remote code intact.
				if ( ! $e->is_retryable() ) {
					$logger->warning(
						sprintf(
							'Not retrying %s after HTTP %d%s: %s',
							$label,
							$e->http_status(),
							'' !== $e->remote_code() ? " ({$e->remote_code()})" : '',
							$e->getMessage()
						),
						array_merge( $attempt_log, [ 'outcome' => 'refused' ] )
					);
					throw $e;
				}

				$last_exception = $e;
				if ( $attempt < $max_attempts ) {
					$delay_us = $this->retry_backoff_us( $attempt );

					// A backoff that outlasts the worker's remaining budget buys nothing, since
					// the retry cannot start before the window shuts. Yield so the next scheduled
					// run gets the full budget.
					if ( ! $this->can_wait( $delay_us ) ) {
						$logger->warning(
							sprintf(
								'Yielding %s at attempt %d/%d rather than sleeping %d ms past the worker deadline: %s',
								$label,
								$attempt,
								$max_attempts,
								(int) round( $delay_us / 1000 ),
								$e->getMessage()
							),
							array_merge( $attempt_log, [ 'outcome' => 'yield' ] )
						);
						throw new DeadlineExceeded( 'Retry backoff would outlast the worker deadline; deferring to the next scheduled chunk.' );
					}

					$logger->warning(
						"Retrying {$label} (attempt {$attempt}/{$max_attempts}): {$e->getMessage()}",
						array_merge( $attempt_log, [ 'outcome' => 'retry', 'backoff_us' => $delay_us ] )
					);
					usleep( $delay_us );
					continue;
				}

				$logger->warning(
					sprintf( 'Gave up on %s after %d/%d attempts: %s', $label, $attempt, $max_attempts, $e->getMessage() ),
					array_merge( $attempt_log, [ 'outcome' => 'exhausted' ] )
				);
			} catch ( RuntimeException $e ) {
				$this->fail_fast_when_out_of_space( $e, $logger, $label );

				$last_exception = $e;
				$attempt_log    = [
					'attempt'      => $attempt,
					'max_attempts' => $max_attempts,
					'reason'       => ErrorReporter::redact_message( $e->getMessage() ),
				];

				if ( $attempt < $max_attempts ) {
					$delay_us = $this->retry_backoff_us( $attempt );

					if ( ! $this->can_wait( $delay_us ) ) {
						$logger->warning(
							sprintf(
								'Yielding %s at attempt %d/%d rather than sleeping %d ms past the worker deadline: %s',
								$label,
								$attempt,
								$max_attempts,
								(int) round( $delay_us / 1000 ),
								$e->getMessage()
							),
							array_merge( $attempt_log, [ 'outcome' => 'yield' ] )
						);
						throw new DeadlineExceeded( 'Retry backoff would outlast the worker deadline; deferring to the next scheduled chunk.' );
					}

					$logger->warning(
						"Retrying {$label} (attempt {$attempt}/{$max_attempts}): {$e->getMessage()}",
						array_merge( $attempt_log, [ 'outcome' => 'retry', 'backoff_us' => $delay_us ] )
					);
					usleep( $delay_us );
					continue;
				}

				$logger->warning(
					sprintf( 'Gave up on %s after %d/%d attempts: %s', $label, $attempt, $max_attempts, $e->getMessage() ),
					array_merge( $attempt_log, [ 'outcome' => 'exhausted' ] )
				);
			}
		}
		throw $last_exception;
	}

	/**
	 * A full destination disk stays full however often the chunk is resent, and a
	 * deferral would restart the attempt count on the next tick and loop for ever.
	 * Ends the retry at once, from either side's message.
	 *
	 * @throws DestinationOutOfSpace When the failure says the database has no room.
	 */
	private function fail_fast_when_out_of_space( RuntimeException $e, MigrationLogger $logger, string $label ): void {
		if ( ! OutOfSpace::matches( $e->getMessage() ) ) {
			return;
		}

		$logger->error(
			sprintf( 'Not retrying %s, the destination database is out of space: %s', $label, $e->getMessage() ),
			[ 'outcome' => 'out_of_space' ]
		);

		throw new DestinationOutOfSpace( $e->getMessage() );
	}

	/**
	 * Backoff before the next attempt: 500 ms, 1 s, 2 s, ...
	 */
	private function retry_backoff_us( int $attempt ): int {
		return 500_000 * (int) pow( 2, max( 0, $attempt - 1 ) );
	}

	/**
	 * Whether a sleep of this length still leaves time for the retry. Waking with
	 * little left gives the request a tiny timeout and a failure the worker cannot
	 * attribute to the link.
	 */
	private function can_wait( int $delay_us ): bool {
		if ( null === $this->active_deadline ) {
			return true;
		}

		$needed = ( $delay_us / 1_000_000 ) + (float) $this->request_timeout();

		return ( microtime( true ) + $needed ) < $this->active_deadline;
	}

	/**
	 * Record what a batch request looked like and how it went. A short timeout on a
	 * large batch looks like a slow link unless the effective window, the bytes on
	 * the wire and the remaining budget are visible, and only this method knows them.
	 *
	 * No paths, data or hashes are recorded. A request id ties a batch to its retries.
	 *
	 * @param string   $outcome     sent | retry | split | refused | exhausted
	 * @param int|null $http_status Status the remote answered with, if any.
	 */
	private function log_batch_request(
		MigrationLogger $logger,
		string $direction,
		string $stage,
		int $raw_bytes,
		int $file_count,
		string $outcome,
		?int $http_status = null,
		?string $request_id = null
	): void {
		$payload = [
			'direction'    => $direction,
			'stage'        => $stage,
			'files'        => $file_count,
			'raw_bytes'    => $raw_bytes,
			// What the host sees. The wire carries base64, so a batch can be refused as
			// "too large" while every number the sender holds says it is small.
			'encoded_bytes' => (int) ceil( $raw_bytes / 3 ) * 4,
			'timeout_s'    => $this->request_timeout(),
			'remaining_s'  => null === $this->active_deadline
				? null
				: max( 0, (int) round( $this->active_deadline - microtime( true ) ) ),
			'outcome'      => $outcome,
		];

		if ( null !== $http_status ) {
			$payload['http_status'] = $http_status;
		}
		if ( null !== $request_id ) {
			$payload['request_id'] = $request_id;
		}

		$logger->info(
			sprintf(
				'Batch request %s: %d file(s), %d raw bytes (%d encoded), %ds timeout.',
				$outcome,
				$file_count,
				$raw_bytes,
				$payload['encoded_bytes'],
				$payload['timeout_s']
			),
			$payload
		);
	}

	private function batch_raw_bytes_for_paths( array $paths, array $entries ): int {
		$bytes = 0;
		foreach ( $paths as $path ) {
			$bytes += (int) ( $entries[ $path ]['size'] ?? 0 );
		}

		return $bytes;
	}

	private function transport_attempt_log( int $attempt, int $max_attempts, HttpRequestException $e ): array {
		$log = [
			'attempt'      => $attempt,
			'max_attempts' => $max_attempts,
			'reason'       => ErrorReporter::redact_message( $e->getMessage() ),
		];

		if ( $e->http_status() > 0 ) {
			$log['http_status'] = $e->http_status();
		}

		if ( '' !== $e->remote_code() ) {
			$log['remote_code'] = $e->remote_code();
		}

		return $log;
	}

	/**
	 * Whether any table in the list has not yet been fully processed, per the
	 * in-memory offsets map (loaded from table_offsets_json).
	 *
	 * @param array<string, array{offset: int, done: bool}> $offsets MD5(table) => progress.
	 * @param string[] $tables Full table names.
	 */
	private function has_pending_from_offsets( array $offsets, array $tables ): bool {
		foreach ( $tables as $table ) {
			$entry = $offsets[ md5( $table ) ] ?? [ 'done' => false ];
			if ( ! ( $entry['done'] ?? false ) ) {
				return true;
			}
		}
		return false;
	}

// phpcs:enable WordPress.Security.EscapeOutput.ExceptionNotEscaped
}