<?php

declare(strict_types=1);

namespace SiteFort\Modules\Scanner\Queue;

use SiteFort\Modules\Scanner\FileIntegrity\FileCollector;
use SiteFort\Modules\Scanner\FileIntegrity\HashService;
use SiteFort\Modules\Scanner\FileIntegrity\FSMEngine;
use SiteFort\Modules\Scanner\Repository\FileStateRepository;
use SiteFort\Modules\Scanner\Repository\FileIndexRepository;
use SiteFort\Modules\Scanner\Cloud\CloudScanService;
use SiteFort\Modules\Scanner\Repository\CloudJobRepository;
use SiteFort\Modules\Scanner\Repository\IssueRepository;
use SiteFort\Modules\Scanner\ScanController;
use SiteFort\Modules\Scanner\Service\ScannerSettings;
use SiteFort\Modules\Scanner\Service\IssueType;
use SiteFort\Modules\Scanner\Queue\CloudResultProcessor;
use SiteFort\Modules\Scanner\Queue\Phase\PhaseServiceInterface;
use SiteFort\Modules\Scanner\Queue\Phase\FileIndexPhaseService;
use SiteFort\Modules\Scanner\Queue\Phase\CloudUploadPhaseService;
use SiteFort\Modules\Scanner\Queue\Phase\SystemChecksPhaseService;
use SiteFort\Modules\Scanner\Queue\Phase\CompletionGatePhaseService;

// If this file is called directly, abort.
if ( ! defined( 'ABSPATH' ) ) {
	exit;
}

/**
 * Coordinates the scanner queue phases for a scan run.
 */
class ScanOrchestrator {
	private const RR_PHASES = [ 'discover', 'hash', 'cloud_check', 'upload', 'poll' ];
	private const MAX_HASH_CHECK_BATCH = 250;
	private const AWAITING_UPLOAD_REQUEUE_AFTER = 300; // seconds
	private const MAX_CLOUD_ERRORS = 3;
	private const MAX_ACTIVE_CLOUD_JOBS = 12;
	private const PARTIAL_UPLOAD_FLUSH_AFTER = 20;
	private const STARTUP_MARKER_SLUG = 'scanner_startup';
	private const CREDIT_EXHAUSTED_STOP_MESSAGE = 'Cloud scan credits are exhausted for this period. Upgrade your plan or wait until credits reset.';
	private const CLOUD_RETRY_DEFER_TRANSIENT_PREFIX = 'sitefort_cloud_retry_defer_until_';

	/**
	 * FSM Engine.
	 *
	 * @var FSMEngine
	 */
	private $fsm;

	/**
	 * File state repository.
	 *
	 * @var FileStateRepository
	 */
	private $file_repository;

	/**
	 * File index repository.
	 *
	 * @var FileIndexRepository
	 */
	private $file_index_repository;

	/**
	 * Cloud job repository.
	 *
	 * @var CloudJobRepository
	 */
	private $job_repository;

	/**
	 * File collector.
	 *
	 * @var FileCollector
	 */
	private $collector;

	/**
	 * Hash service (for hashing phase only).
	 *
	 * @var HashService
	 */
	private $hash_service;

	/**
	 * Cloud scan service
	 *
	 * @var CloudScanService
	 */
	private $cloud_service;

	/**
	 * Current scan ID.
	 *
	 * @var string
	 */
	private $scan_id;

	/**
	 * Scan Controller.
	 *
	 * @var ScanController|null
	 */
	private $controller;

	/**
	 * Current cloud job ID
	 *
	 * @var string|null
	 */
	private $current_job_id;

	/**
	 * Cached scan start time (epoch seconds).
	 *
	 * @var int|null
	 */
	private $scan_start_time;

	/**
	 * Cached scan type (standard|deep), lazily resolved from the queue.
	 *
	 * @var string|null
	 */
	private $scan_type;

	/**
	 * Cached scan max duration (seconds).
	 *
	 * @var int|null
	 */
	private $scan_max_duration;

	/**
	 * Cached stage execution limit (seconds).
	 *
	 * @var int|null
	 */
	private $stage_execution_limit;

	/**
	 * Whether inline orchestrator system checks are enabled.
	 *
	 * @var bool
	 */
	private bool $inline_system_checks_enabled = true;

	/**
	 * AIMD Controller.
	 *
	 * @var \SiteFort\Modules\Scanner\Service\AIMDController
	 */
	private $aimd;

	/**
	 * Cloud result processor (extracted responsibility).
	 *
	 * @var CloudResultProcessor
	 */
	private CloudResultProcessor $result_processor;

	/**
	 * Last pressure level that was logged (transition-only logging).
	 *
	 * @var string|null
	 */
	private static $last_logged_pressure_level = null;

	// --- Per-tick in-memory caches to reduce transient/DB round-trips ---
	/** @var string|null RR phase cached in memory; persisted to transient once per tick. */
	private $rr_phase_cache = null;
	/** @var bool|null Once true, file index stays complete for the rest of the tick. */
	private $file_index_complete_cache = null;
	/** @var bool|null Hashing-complete result cached with TTL. */
	private $hashing_complete_cache = null;
	/** @var int Timestamp of last hashing-complete DB check. */
	private $hashing_complete_cache_time = 0;
	/** @var array|null Cached ScanQueue::get() result with TTL. */
	private $queue_cache = null;
	/** @var int Timestamp of last ScanQueue::get() call. */
	private $queue_cache_time = 0;
	/** @var int Epoch timestamp when cloud hash checks may resume. */
	private int $cloud_retry_defer_until = 0;

	/**
	 * Constructor.
	 *
	 * @param string $scan_id Scan UUID.
	 */
	public function __construct( string $scan_id, ?ScanController $controller = null ) {
		$this->scan_id         = $scan_id;
		$this->controller      = $controller;
		$this->fsm             = new FSMEngine();
		$this->file_repository = new FileStateRepository();
		$this->file_index_repository = new FileIndexRepository();
		$this->job_repository  = new CloudJobRepository();
		$this->collector       = new FileCollector();
		$this->hash_service    = new HashService();
		$this->cloud_service   = new CloudScanService( $scan_id );
		$this->current_job_id  = $this->get_active_job_id();

		// Initialize AIMD Controller
		if ( ! class_exists( 'SiteFort\\Modules\\Scanner\\Service\\AIMDController' ) ) {
			require_once __DIR__ . '/../Service/AIMDController.php';
		}
		$this->aimd             = new \SiteFort\Modules\Scanner\Service\AIMDController( $scan_id );
		$this->result_processor = new CloudResultProcessor( $scan_id, $this->fsm );
	}

	/**
	 * Override the per-slice execution limit.
	 *
	 * @param int $seconds
	 * @return void
	 */
	public function set_execution_limit( int $seconds ): void {
		$this->stage_execution_limit = max( 1, $seconds );
	}

	/**
	 * Enable or disable inline system checks during orchestrator execution.
	 *
	 * @param bool $enabled
	 * @return void
	 */
	public function set_inline_system_checks_enabled( bool $enabled ): void {
		$this->inline_system_checks_enabled = $enabled;
	}

	/**
	 * Earliest time the next file worker should run.
	 *
	 * Local discovery and hashing may continue immediately. Pure cloud-check
	 * retries wait for the persisted backoff window to avoid counting local
	 * backoff as repeated cloud failures.
	 *
	 * @return int Unix timestamp.
	 */
	public function get_next_file_worker_available_at(): int {
		if ( $this->has_local_file_worker_work() ) {
			return time();
		}

		$defer_until = $this->get_cloud_retry_defer_until();
		return $defer_until > time() ? $defer_until : time();
	}

	/**
	 * Run the FSM-based scan orchestration (AIMD Execution Gate).
	 * This is called by ScanWorker for each batch iteration.
	 *
	 * @return array Orchestration result with progress stats.
	 */
	public function run(): array {
		// Feature flag: route to phase-service architecture when enabled.
		if ( get_option( 'sitefort_phase_services_enabled' ) ) {
			return $this->run_v2();
		}

		$stats = $this->build_stats();

		if ( $this->handle_upload_failure_abort() ) {
			return $stats;
		}

		if ( $this->should_stop_for_scan_duration() ) {
			$this->fail_scan_due_to_timeout();
			return $stats;
		}

		if ( ! $this->run_startup_preflight() ) {
			return $stats;
		}

		// Phase 1.5: System Checks
		// Run at most one check per orchestrator cycle so polling UI can
		// render intermediate "active" states instead of jumping pending->complete.
		// Runs BEFORE the timer is set so a slow system check (e.g. reputation) does
		// not eat into the AIMD file-scanning budget.
		if ( $this->inline_system_checks_enabled ) {
			$this->run_system_checks_phase();
			$queue_after_checks = ScanQueue::get();
			if ( ! $queue_after_checks || ( $queue_after_checks['status'] ?? '' ) === 'failed' ) {
				return $stats;
			}
		}

		// --- TIMER SETUP (after system checks so budget is fully available for file scanning) ---
		$start_time   = time();
		$exec_budget  = $this->get_stage_execution_budget();

		// --- AIMD EXECUTION GATE ---
		$this->aimd->evaluate_aimd_cycle();

		$last_heartbeat_at = $start_time;

		while ( ( time() - $start_time ) < $exec_budget ) {
			if ( $this->should_stop_for_scan_duration() ) {
				$this->fail_scan_due_to_timeout();
				return $stats;
			}

			// Mid-tick heartbeat: update last_ping every 20s so the watchdog
			// does not misclassify a slow-but-live tick as a stalled scan.
			if ( ( time() - $last_heartbeat_at ) >= 20 ) {
				ScanQueue::heartbeat();
				$last_heartbeat_at = time();
			}

			$did_work = false;

			$phase   = $this->get_rr_phase();
			$attempt = 0;

			while ( $attempt < count( self::RR_PHASES ) ) {
				switch ( $phase ) {
					case 'discover':
						$did_work = $this->run_discovery_phase_rr( $stats );
						break;
					case 'poll':
						$did_work = $this->run_poll_phase( $stats );
						break;
					case 'cloud_check':
						$did_work = $this->run_cloud_check_phase_rr( $stats );
						break;
					case 'upload':
						$did_work = $this->run_upload_phase_rr( $stats );
						break;
					case 'hash':
					default:
						$did_work = $this->run_hash_phase_rr( $stats );
						break;
				}

				$phase = $this->advance_rr_phase( $phase );
				$this->set_rr_phase( $phase );

				if ( $did_work ) {
					$this->persist_rr_phase();
					break;
				}

				$attempt++;
			}

			// Check Completion
			if ( $this->check_completion() ) {
				$stats['complete'] = true;
				break; // Scan done
			}
			
			// If the scan was marked failed, stop the loop to avoid running forever.
			$queue = $this->get_cached_queue();
			if ( $queue && ( $queue['status'] ?? '' ) === 'failed' ) {
				break;
			}

			// No file work this iteration.
			if ( ! $did_work ) {
				if ( $this->check_completion() ) {
					$stats['complete'] = true;
					break;
				}
				// File pipeline is idle (waiting for cloud, or fully done) but system
				// checks are still pending. Run the next one and stay in the loop so
				// we never exit with work unfinished and create a gap that needs recovery.
				if ( $this->inline_system_checks_enabled && ! $this->are_system_checks_complete() ) {
					$this->run_system_checks_phase();
					$this->reset_queue_cache();
					continue;
				}
				break;
			}
		}

		$this->persist_rr_phase();
		return $stats;
	}

	// =========================================================================
	// PHASE SERVICE ARCHITECTURE (v2 - enabled via sitefort_phase_services_enabled option)
	// =========================================================================

	/**
	 * Phase-service based scan execution.
	 *
	 * Identical control flow to run() but:
	 * - RR phase state is persisted to DB (component_status) instead of transients.
	 * - Phase logic is delegated to PhaseServiceInterface instances.
	 * - On first invocation migrates any existing transient phase to DB.
	 *
	 * @return array Orchestration result with progress stats.
	 */
	private function run_v2(): array {
		$stats = $this->build_stats();

		if ( $this->handle_upload_failure_abort() ) {
			return $stats;
		}

		if ( $this->should_stop_for_scan_duration() ) {
			$this->fail_scan_due_to_timeout();
			return $stats;
		}

		if ( ! $this->run_startup_preflight() ) {
			return $stats;
		}

		// Build phase services backed by closures into existing private methods.
		$fileIndex = new FileIndexPhaseService(
			fn( array &$s ) => $this->run_discovery_phase_rr( $s ),
			fn() => $this->is_file_index_complete()
		);

		$cloudUpload = new CloudUploadPhaseService(
			fn( array &$s ) => $this->run_hash_phase_rr( $s ),
			fn( array &$s ) => $this->run_cloud_check_phase_rr( $s ),
			fn( array &$s ) => $this->run_upload_phase_rr( $s ),
			fn() => $this->check_hashing_complete()
		);

		$systemChecks = new SystemChecksPhaseService(
			fn( array &$s ) => $this->run_system_checks_phase(),
			fn() => $this->are_system_checks_complete()
		);

		$completionGate = new CompletionGatePhaseService(
			fn( array &$s ) => $this->run_poll_phase( $s ),
			fn() => $this->check_completion()
		);

		/** @var PhaseServiceInterface[] RR participants (system checks run separately) */
		$rr_phases = [ $fileIndex, $cloudUpload, $completionGate ];
		$rr_phase_names = array_map( fn( PhaseServiceInterface $p ) => $p->get_phase_name(), $rr_phases );

		// System checks always first, outside RR loop (same as run()).
		if ( $this->inline_system_checks_enabled ) {
			$systemChecks->tick( $stats );
			$queue_after_checks = ScanQueue::get();
			if ( ! $queue_after_checks || ( $queue_after_checks['status'] ?? '' ) === 'failed' ) {
				return $stats;
			}
		}

		// Timer + AIMD (identical to run())
		$start_time  = time();
		$exec_budget = $this->get_stage_execution_budget();
		$this->aimd->evaluate_aimd_cycle();
		$last_heartbeat_at = $start_time;

		// Migrate legacy transient phase to DB on first v2 invocation.
		$this->migrate_rr_phase_to_db( $rr_phase_names );

		while ( ( time() - $start_time ) < $exec_budget ) {
			if ( $this->should_stop_for_scan_duration() ) {
				$this->fail_scan_due_to_timeout();
				return $stats;
			}

			if ( ( time() - $last_heartbeat_at ) >= 20 ) {
				ScanQueue::heartbeat();
				$last_heartbeat_at = time();
			}

			$did_work = false;
			$phase    = $this->get_rr_phase_from_db( $rr_phase_names );

			for ( $attempt = 0; $attempt < count( $rr_phases ); $attempt++ ) {
				$service = $this->find_phase_service_by_name( $rr_phases, $phase );
				if ( $service ) {
					$did_work = $service->tick( $stats );
				}
				$phase = $this->advance_rr_phase_v2( $phase, $rr_phase_names );
				$this->persist_rr_phase_to_db( $phase );
				if ( $did_work ) {
					break;
				}
			}

			if ( $completionGate->is_complete() ) {
				$stats['complete'] = true;
				break;
			}

			$queue = $this->get_cached_queue();
			if ( $queue && ( $queue['status'] ?? '' ) === 'failed' ) {
				break;
			}

			if ( ! $did_work ) {
				if ( $completionGate->is_complete() ) {
					$stats['complete'] = true;
					break;
				}
				if ( $this->inline_system_checks_enabled && ! $systemChecks->is_complete() ) {
					$systemChecks->tick( $stats );
					$this->reset_queue_cache();
					continue;
				}
				break;
			}
		}

		return $stats;
	}

	/**
	 * Run one bounded task-queue slice for discovery, hashing, and cloud hash checks.
	 *
	 * Uploading, cloud polling, and system checks are handled by dedicated tasks.
	 *
	 * @return array<string,mixed>
	 */
	public function run_task_queue_scan_worker_slice(): array {
		$stats = $this->build_stats();

		if ( $this->handle_upload_failure_abort() ) {
			return $stats;
		}

		if ( $this->should_stop_for_scan_duration() ) {
			$this->fail_scan_due_to_timeout();
			return $stats;
		}

		if ( ! $this->run_startup_preflight() ) {
			return $stats;
		}

		$allowed_phases = [ 'discover', 'hash', 'cloud_check' ];
		$start_time = time();
		$exec_budget = $this->get_stage_execution_budget();
		$this->aimd->evaluate_aimd_cycle();
		$last_heartbeat_at = $start_time;

		while ( ( time() - $start_time ) < $exec_budget ) {
			if ( $this->should_stop_for_scan_duration() ) {
				$this->fail_scan_due_to_timeout();
				return $stats;
			}

			if ( ( time() - $last_heartbeat_at ) >= 20 ) {
				ScanQueue::heartbeat();
				$last_heartbeat_at = time();
			}

			$did_work = false;
			$phase = $this->get_rr_phase();
			if ( ! in_array( $phase, $allowed_phases, true ) ) {
				$phase = $allowed_phases[0];
			}

			for ( $attempt = 0; $attempt < count( $allowed_phases ); $attempt++ ) {
				switch ( $phase ) {
					case 'discover':
						$did_work = $this->run_discovery_phase_rr( $stats );
						break;
					case 'cloud_check':
						$did_work = $this->run_cloud_check_phase_rr( $stats );
						break;
					case 'hash':
					default:
						$did_work = $this->run_hash_phase_rr( $stats );
						break;
				}

				$phase = $this->advance_rr_phase_for_set( $phase, $allowed_phases );
				$this->set_rr_phase( $phase );

				if ( $did_work ) {
					$this->persist_rr_phase();
					break;
				}
			}

			if ( $this->is_current_scan_terminal() ) {
				break;
			}

			if ( ! $did_work ) {
				break;
			}
		}

		$this->persist_rr_phase();
		return $stats;
	}

	/**
	 * Build the default stats payload for a scanner slice.
	 *
	 * @return array<string,mixed>
	 */
	private function build_stats(): array {
		return [
			'discovered'   => 0,
			'hashed'       => 0,
			'cloud_checks' => 0,
			'uploaded'     => 0,
			'scanned'      => 0,
			'verified'     => $this->get_verified_count(),
			'complete'     => false,
		];
	}

	/**
	 * Fail the scan when upload retries exceeded the AIMD threshold.
	 *
	 * @return bool True when the scan was aborted.
	 */
	private function handle_upload_failure_abort(): bool {
		if ( ! $this->aimd->should_abort_scan() ) {
			return false;
		}

		$msg = 'Scan aborted due to persistent upload failures. Please check your connection or server logs.';

		if ( class_exists( '\SiteFort\Modules\Scanner\Service\ScanAlertService' ) ) {
			\SiteFort\Modules\Scanner\Service\ScanAlertService::send_alert(
				'upload_failure',
				$msg,
				[ 'failures' => $this->aimd->get_failure_count() ]
			);
		} else {
			sitefort_log( '[SiteFort] ' . $msg );
		}

		if ( class_exists( '\SiteFort\Modules\Scanner\Service\ScanLogger' ) ) {
			\SiteFort\Modules\Scanner\Service\ScanLogger::log( $msg, 3, 'error' );
		}

		CloudResultProcessor::stage_skipped_issue(
			'(scan)',
			'Scan aborted due to persistent upload failures.',
			[ 'reason' => 'upload_failures_exceeded' ]
		);

		ScanQueue::fail(
			$msg,
			[
				'error_code'        => 'upload_failures_exceeded',
				'stage'             => 'upload',
				'retryable'         => true,
				'status_code'       => 503,
				'technical_message' => sprintf( 'AIMD abort after %d upload failures', (int) $this->aimd->get_failure_count() ),
			]
		);

		return true;
	}

	/**
	 * Run one-time startup checks required before the scanner loop starts.
	 *
	 * @return bool True when startup preflight passed.
	 */
	private function run_startup_preflight(): bool {
		if ( $this->has_startup_logged() ) {
			return true;
		}

		$this->aimd->log_startup_state();

		$disk_free = @disk_free_space( ABSPATH );
		if ( false !== $disk_free && $disk_free < 50 * 1024 * 1024 ) {
			$disk_mb = (int) floor( $disk_free / ( 1024 * 1024 ) );
			$msg     = "Scan aborted: only {$disk_mb}MB disk space available. Minimum 50MB required.";
			if ( class_exists( '\SiteFort\Modules\Scanner\Service\ScanLogger' ) ) {
				\SiteFort\Modules\Scanner\Service\ScanLogger::log( $msg, 3, 'error' );
			}
			ScanQueue::fail(
				$msg,
				[
					'error_code'  => 'insufficient_disk_space',
					'stage'       => 'preflight',
					'retryable'   => false,
					'status_code' => 507,
				]
			);
			return false;
		}

		$this->mark_startup_logged();
		return true;
	}

	/**
	 * True when discovery/hash/check-hash still have work remaining.
	 *
	 * @return bool
	 */
	public function has_file_worker_work(): bool {
		if ( ! $this->is_file_index_complete() ) {
			return true;
		}

		return $this->fsm->has_files_in_states(
			$this->scan_id,
			[ 'discovered', 'hashing', 'hash_complete', 'cloud_hash_check' ]
		);
	}

	/**
	 * Determine whether an upload batch is ready to run now.
	 *
	 * @param array<string,mixed> $payload
	 * @return bool
	 */
	public function has_ready_upload_batch( array $payload = [] ): bool {
		$metrics = $this->aimd->sense_resources();
		$pressure = $this->aimd->get_pressure_state_from_metrics( $metrics );
		if ( $this->aimd->get_upload_window_for_phase( $pressure, $metrics ) <= 0 ) {
			return false;
		}

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

		$hashing_complete = $this->check_hashing_complete();
		$files = $this->select_upload_candidates(
			$hashing_complete,
			(int) ( $payload['max_files'] ?? 0 ),
			$this->resolve_upload_max_size_bytes( $payload )
		);

		return ! empty( $files );
	}

	/**
	 * Run one upload batch task.
	 *
	 * @param array<string,mixed> $payload
	 * @return array<string,mixed>
	 */
	public function run_upload_batch_task( array $payload = [] ): array {
		$metrics = $this->aimd->sense_resources();
		$pressure = $this->aimd->get_pressure_state_from_metrics( $metrics );
		if ( $this->aimd->get_upload_window_for_phase( $pressure, $metrics ) <= 0 ) {
			return [
				'uploaded_count' => 0,
				'job_id'         => null,
				'has_more'       => false,
				'deferred'       => true,
			];
		}

		if ( ! $this->has_cloud_upload_capacity() ) {
			return [
				'uploaded_count' => 0,
				'job_id'         => null,
				'has_more'       => false,
				'deferred'       => true,
			];
		}

		$hashing_complete = $this->check_hashing_complete();
		$max_files = (int) ( $payload['max_files'] ?? 0 );
		$max_size_bytes = $this->resolve_upload_max_size_bytes( $payload );
		$files = $this->select_upload_candidates( $hashing_complete, $max_files, $max_size_bytes );

		if ( empty( $files ) ) {
			return [
				'uploaded_count' => 0,
				'job_id'         => null,
				'has_more'       => false,
			];
		}

		$previous_job_id = $this->current_job_id;
		$uploaded_count = $this->run_upload_phase( $files );
		$job_id = null;
		if ( $uploaded_count > 0 && ! empty( $this->current_job_id ) && $this->current_job_id !== $previous_job_id ) {
			$job_id = $this->current_job_id;
		}

		return [
			'uploaded_count' => $uploaded_count,
			'job_id'         => $job_id,
			'has_more'       => $this->has_cloud_upload_capacity() && ! empty(
				$this->select_upload_candidates(
					$this->check_hashing_complete(),
					$max_files,
					$max_size_bytes
				)
			),
		];
	}

	/**
	 * Poll a specific set of cloud job IDs.
	 *
	 * @param array<int,string> $job_ids
	 * @return array<string,mixed>|\WP_Error
	 */
	public function poll_job_ids( array $job_ids ) {
		$job_ids = array_values(
			array_filter(
				array_unique(
					array_map(
						static fn( $job_id ): string => sanitize_text_field( (string) $job_id ),
						$job_ids
					)
				)
			)
		);

		if ( empty( $job_ids ) ) {
			return [
				'processed'             => 0,
				'pending_job_ids'       => [],
				'completed_job_ids'     => [],
				'failed_job_ids'        => [],
				'requeued_upload'       => false,
				'youngest_pending_age' => null,
			];
		}

		$jobs = [];
		foreach ( $job_ids as $job_id ) {
			$job = $this->job_repository->get_by_job_id( $job_id );
			if ( is_array( $job ) && (string) ( $job['scan_id'] ?? '' ) === $this->scan_id ) {
				$jobs[] = $job;
			}
		}

		if ( empty( $jobs ) ) {
			return [
				'processed'             => 0,
				'pending_job_ids'       => [],
				'completed_job_ids'     => [],
				'failed_job_ids'        => $job_ids,
				'requeued_upload'       => false,
				'youngest_pending_age' => null,
			];
		}

		return $this->process_polled_jobs( $jobs );
	}

	/**
	 * Poll active cloud jobs for this scan.
	 *
	 * @return array<string,mixed>|\WP_Error
	 */
	public function poll_active_jobs() {
		$jobs = $this->job_repository->get_active_jobs_for_scan( $this->scan_id, 50 );
		if ( empty( $jobs ) ) {
			return [
				'processed'             => 0,
				'pending_job_ids'       => [],
				'completed_job_ids'     => [],
				'failed_job_ids'        => [],
				'requeued_upload'       => false,
				'youngest_pending_age' => null,
			];
		}

		return $this->process_polled_jobs( $jobs );
	}

	/**
	 * Finalize gate for the task queue path.
	 *
	 * @return bool
	 */
	public function is_ready_for_finalize(): bool {
		return $this->check_completion();
	}

	/**
	 * Expose system-check completion to task handlers.
	 *
	 * @return bool
	 */
	public function are_all_system_checks_complete(): bool {
		return $this->are_system_checks_complete();
	}

	/**
	 * Recover stale active-scan runtime state for the durable task queue path.
	 *
	 * This is intentionally invoked under the scan-row CAS lease so stale upload
	 * recovery and file/cloud reconciliation do not race with active tasks.
	 *
	 * @return array<string,mixed>
	 */
	public function recover_task_queue_runtime_state(): array {
		return [
			'stale_uploads'          => $this->recover_stale_uploading_files(),
			'cloud_states_recovered' => $this->reconcile_completed_jobs(),
		];
	}

	/**
	 * Get the current RR phase from DB-backed component_status.
	 *
	 * @param string[] $valid_phases
	 * @return string
	 */
	private function get_rr_phase_from_db( array $valid_phases ): string {
		if ( $this->rr_phase_cache !== null && in_array( $this->rr_phase_cache, $valid_phases, true ) ) {
			return $this->rr_phase_cache;
		}
		$queue = $this->get_cached_queue();
		$component_status = ( $queue['component_status'] ?? [] );
		if ( ! is_array( $component_status ) ) {
			$component_status = [];
		}
		$phase = (string) ( $component_status['rr_phase'] ?? '' );
		if ( ! in_array( $phase, $valid_phases, true ) ) {
			$phase = $valid_phases[0];
		}
		$this->rr_phase_cache = $phase;
		return $phase;
	}

	/**
	 * Persist the current RR phase to DB-backed component_status.
	 *
	 * @param string $phase
	 */
	private function persist_rr_phase_to_db( string $phase ): void {
		$this->rr_phase_cache = $phase;
		ScanQueue::set_component_status( 'rr_phase', $phase );
		$this->reset_queue_cache();
	}

	/**
	 * Advance to the next phase in the v2 phase list.
	 *
	 * @param string   $current
	 * @param string[] $phase_names
	 * @return string
	 */
	private function advance_rr_phase_v2( string $current, array $phase_names ): string {
		$idx = array_search( $current, $phase_names, true );
		if ( $idx === false ) {
			return $phase_names[0];
		}
		return $phase_names[ ( $idx + 1 ) % count( $phase_names ) ];
	}

	/**
	 * Find a phase service by its phase name.
	 *
	 * @param PhaseServiceInterface[] $services
	 * @param string                  $phase_name
	 * @return PhaseServiceInterface|null
	 */
	private function find_phase_service_by_name( array $services, string $phase_name ): ?PhaseServiceInterface {
		foreach ( $services as $service ) {
			if ( $service->get_phase_name() === $phase_name ) {
				return $service;
			}
		}
		return null;
	}

	/**
	 * On first v2 invocation, migrate any existing transient phase to DB.
	 * Prevents scans in-flight at deploy time from restarting from discover.
	 *
	 * @param string[] $valid_phases
	 */
	private function migrate_rr_phase_to_db( array $valid_phases ): void {
		$queue = $this->get_cached_queue();
		$component_status = ( $queue['component_status'] ?? [] );
		if ( is_array( $component_status ) && isset( $component_status['rr_phase'] ) ) {
			// Already migrated.
			return;
		}

		$transient_phase = get_transient( 'sitefort_rr_phase_' . $this->scan_id );
		$phase = ( is_string( $transient_phase ) && in_array( $transient_phase, $valid_phases, true ) )
			? $transient_phase
			: $valid_phases[0];
		$this->persist_rr_phase_to_db( $phase );
	}

	// =========================================================================
	// END PHASE SERVICE ARCHITECTURE (v2)
	// =========================================================================

	/**
	 * Round-robin phase helpers.
	 */
	private function get_rr_phase(): string {
		if ( $this->rr_phase_cache !== null ) {
			return $this->rr_phase_cache;
		}
		$phase = get_transient( 'sitefort_rr_phase_' . $this->scan_id );
		if ( ! in_array( $phase, self::RR_PHASES, true ) ) {
			$phase = self::RR_PHASES[0];
		}
		$this->rr_phase_cache = $phase;
		return $phase;
	}

	private function set_rr_phase( string $phase ): void {
		$this->rr_phase_cache = $phase;
	}

	private function persist_rr_phase(): void {
		if ( $this->rr_phase_cache !== null ) {
			set_transient( 'sitefort_rr_phase_' . $this->scan_id, $this->rr_phase_cache, HOUR_IN_SECONDS );
		}
	}

	private function get_cached_queue(): ?array {
		$now = time();
		if ( $this->queue_cache !== null && ( $now - $this->queue_cache_time ) < 3 ) {
			return $this->queue_cache;
		}
		$this->queue_cache      = ScanQueue::get();
		$this->queue_cache_time = $now;
		return $this->queue_cache;
	}

	/**
	 * Invalidate local queue cache so follow-up reads observe latest DB state.
	 *
	 * @return void
	 */
	private function reset_queue_cache(): void {
		$this->queue_cache = null;
		$this->queue_cache_time = 0;
	}

	/**
	 * Determine whether startup state was already logged for this scan.
	 *
	 * Uses queue component_status (durable DB state) instead of transients.
	 *
	 * @return bool
	 */
	private function has_startup_logged(): bool {
		$queue = $this->get_cached_queue();
		if ( ! $queue ) {
			return false;
		}

		$stored = $queue['component_status'] ?? [];
		if ( ! is_array( $stored ) ) {
			return false;
		}

		$entry = $stored[ self::STARTUP_MARKER_SLUG ] ?? null;
		$status = '';
		if ( is_array( $entry ) ) {
			$status = (string) ( $entry['status'] ?? '' );
		} elseif ( is_string( $entry ) ) {
			$status = $entry;
		}

		return $status === 'complete';
	}

	/**
	 * Persist startup marker for this scan.
	 *
	 * @return void
	 */
	private function mark_startup_logged(): void {
		ScanQueue::set_component_status( self::STARTUP_MARKER_SLUG, 'complete' );
		$this->reset_queue_cache();
	}

	private function advance_rr_phase( string $current ): string {
		$idx = array_search( $current, self::RR_PHASES, true );
		if ( $idx === false ) {
			return self::RR_PHASES[0];
		}
		$next = ( $idx + 1 ) % count( self::RR_PHASES );
		return self::RR_PHASES[ $next ];
	}

	/**
	 * Advance within a subset of RR phases.
	 *
	 * @param string   $current
	 * @param string[] $phases
	 * @return string
	 */
	private function advance_rr_phase_for_set( string $current, array $phases ): string {
		$idx = array_search( $current, $phases, true );
		if ( false === $idx ) {
			return $phases[0];
		}

		return $phases[ ( $idx + 1 ) % count( $phases ) ];
	}

	/**
	 * Requeue upload tasks left in-flight after a worker crash or timeout.
	 *
	 * @param int $stuck_after
	 * @param int $limit
	 * @return int
	 */
	private function recover_stale_uploading_files( int $stuck_after = 300, int $limit = 100 ): int {
		global $wpdb;

		$cutoff_time = time() - max( 60, $stuck_after );
		$limit       = max( 1, $limit );
		$table       = $this->table_identifier( $wpdb->prefix . 'sitefort_file_states' );

		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,PluginCheck.Security.DirectDB.UnescapedDBParameter -- Uses a validated plugin-owned table identifier and prepared values.
		$files = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id, state, upload_attempt, file_path
				FROM {$table}
				WHERE scan_id = %s
				AND state = 'uploading'
				AND updated_at < %d
				ORDER BY updated_at ASC
				LIMIT %d",
				$this->scan_id,
				$cutoff_time,
				$limit
			),
			ARRAY_A
		);
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,PluginCheck.Security.DirectDB.UnescapedDBParameter

		if ( empty( $files ) ) {
			return 0;
		}

		$recovered = 0;

		foreach ( $files as $file ) {
			$attempts = (int) ( $file['upload_attempt'] ?? 0 );

			if ( $attempts < 3 ) {
				$transition = $this->fsm->transition(
					(int) $file['id'],
					'uploading',
					'pending_upload',
					[
						'cloud_job_id'  => null,
						'error_message' => 'Upload timeout - retrying',
					]
				);
			} else {
				$transition = $this->fsm->transition(
					(int) $file['id'],
					'uploading',
					'error',
					[
						'cloud_job_id'  => null,
						'error_code'    => 'upload_timeout',
						'error_message' => 'Upload stuck for >5 minutes',
					]
				);

				if ( ! is_wp_error( $transition ) ) {
					CloudResultProcessor::stage_skipped_issue(
						(string) ( $file['file_path'] ?? '' ),
						'Upload failed: cloud upload did not complete within the recovery window',
						[ 'error_code' => 'upload_timeout', 'upload_attempts' => $attempts ]
					);
				}
			}

			if ( ! is_wp_error( $transition ) ) {
				$recovered++;
			}
		}

		return $recovered;
	}

	/**
	 * Round-robin phases: one task per loop.
	 */
	private function run_discovery_phase_rr( array &$stats ): bool {
		if ( $this->is_file_index_complete() ) {
			return false;
		}

		$batch_limit = max( 1, (int) $this->aimd->get_discovery_window_for_phase() );

		$discovered = $this->run_discovery_phase( $batch_limit );

		if ( $discovered > 0 ) {
			$stats['discovered'] += $discovered;
			return true;
		}

		return false;
	}

	private function run_poll_phase( array &$stats ): bool {
		$poll_interval = $this->aimd->get_poll_interval_for_phase();
		$now = time();

		$last_poll_global = (int) get_transient( 'sitefort_last_poll_check_' . $this->scan_id );
		if ( ( $now - $last_poll_global ) < $poll_interval ) {
			return false;
		}

		$polled = $this->poll_results();
		if ( $polled > 0 ) {
			$stats['scanned'] += $polled;
		}

		$reconciled = $this->reconcile_completed_jobs();

		set_transient( 'sitefort_last_poll_check_' . $this->scan_id, $now, 45 );
		return ( $polled > 0 ) || $reconciled;
	}

	private function run_upload_phase_rr( array &$stats ): bool {
		// Cloud credits exhausted - skip uploads.
		if ( get_transient( 'sitefort_credit_exhausted_' . $this->scan_id ) ) {
			return false;
		}

		$metrics       = $this->aimd->sense_resources();
		$pressure      = $this->aimd->get_pressure_state_from_metrics( $metrics );

		// AIMD controls window: returns MIN_UPLOAD_WINDOW under critical, not 0.
		$upload_window = $this->aimd->get_upload_window_for_phase( $pressure, $metrics );
		if ( $upload_window <= 0 ) {
			return false;
		}

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

		$hashing_complete = $this->check_hashing_complete();
		$files = $this->select_upload_candidates( $hashing_complete );
		if ( empty( $files ) ) {
			return false;
		}

		$uploaded = $this->run_upload_phase( $files );
		if ( $uploaded > 0 ) {
			$stats['uploaded'] += $uploaded;
			return true;
		}

		return false;
	}

	private function run_hash_phase_rr( array &$stats ): bool {
		$metrics     = $this->aimd->sense_resources();
		$pressure    = $this->aimd->get_pressure_state_from_metrics( $metrics );
		$buffer_size = $metrics['buffer_size'];

		$this->log_pressure_state( $pressure, $metrics );

		// Stop hashing if all files are hashed or upload buffer is backed up.
		if ( $this->check_hashing_complete() || $buffer_size >= 20 * 1024 * 1024 ) {
			return false;
		}

		// AIMD controls window: returns MIN_HASH_WINDOW under critical, not 0.
		$hashed = $this->run_hashing_phase( $pressure, $metrics );
		if ( $hashed > 0 ) {
			$stats['hashed'] += $hashed;
			$this->log_scan_progress( 'hash' );
			return true;
		}

		return false;
	}

	private function cloud_retry_defer_key(): string {
		return self::CLOUD_RETRY_DEFER_TRANSIENT_PREFIX . $this->scan_id;
	}

	private function get_cloud_retry_defer_until(): int {
		if ( $this->cloud_retry_defer_until > time() ) {
			return $this->cloud_retry_defer_until;
		}

		$stored = get_transient( $this->cloud_retry_defer_key() );
		if ( is_numeric( $stored ) ) {
			$this->cloud_retry_defer_until = (int) $stored;
			return $this->cloud_retry_defer_until;
		}

		$this->cloud_retry_defer_until = 0;
		return 0;
	}

	private function is_cloud_retry_deferred(): bool {
		return $this->get_cloud_retry_defer_until() > time();
	}

	private function defer_cloud_retry( \WP_Error $error, int $error_count = 0 ): void {
		$retry_after = $this->extract_retry_after( $error );
		if ( $retry_after <= 0 ) {
			$attempt = max( 1, $error_count );
			$retry_after = (int) min( 60, 4 * pow( 2, $attempt - 1 ) );
		}

		$retry_after = max( 1, min( 300, $retry_after ) );
		$defer_until = time() + $retry_after;

		$this->cloud_retry_defer_until = $defer_until;
		set_transient( $this->cloud_retry_defer_key(), $defer_until, $retry_after );
	}

	private function clear_cloud_retry_defer(): void {
		$this->cloud_retry_defer_until = 0;
		delete_transient( $this->cloud_retry_defer_key() );
	}

	private function extract_retry_after( \WP_Error $error ): int {
		$error_data = $error->get_error_data( 'api_error' );
		if ( empty( $error_data ) ) {
			$error_data = $error->get_error_data();
		}
		if ( ! is_array( $error_data ) ) {
			return 0;
		}

		if ( isset( $error_data['retry_after'] ) ) {
			return absint( $error_data['retry_after'] );
		}

		$detail = is_array( $error_data['detail'] ?? null ) ? $error_data['detail'] : [];
		if ( isset( $detail['retry_after'] ) ) {
			return absint( $detail['retry_after'] );
		}

		return 0;
	}

	private function has_local_file_worker_work(): bool {
		if ( ! $this->is_file_index_complete() ) {
			return true;
		}

		return $this->fsm->has_files_in_states(
			$this->scan_id,
			[ 'discovered', 'hashing' ]
		);
	}

	private function is_current_scan_terminal(): bool {
		$queue = ScanQueue::get_job( $this->scan_id );
		return is_array( $queue ) && ! in_array( (string) ( $queue['status'] ?? '' ), [ 'pending', 'running' ], true );
	}

	private function run_cloud_check_phase_rr( array &$stats ): bool {
		// Cloud credits exhausted - skip cloud API calls.
		if ( get_transient( 'sitefort_credit_exhausted_' . $this->scan_id ) ) {
			return false;
		}

		if ( $this->is_cloud_retry_deferred() ) {
			return false;
		}

		// No pressure gate: cloud hash check is a network call with negligible local cost.
		$metrics  = $this->aimd->sense_resources();
		$pressure = $this->aimd->get_pressure_state_from_metrics( $metrics );

		$checked = $this->run_cloud_check_phase( $pressure, $metrics );
		if ( $checked > 0 ) {
			$stats['cloud_checks'] += $checked;
			return true;
		}

		return false;
	}

	/**
	 * Phase 1: File Indexing (Discovery).
	 * Discover new files and insert into FSM as 'discovered'.
	 * Supports incremental indexing via batch_limit.
	 *
	 * @param int $batch_limit Limit to keep indexing bounded.
	 * @return int Number of files discovered in this batch.
	 */
	private function run_discovery_phase( int $batch_limit ): int {
		// Check if file index is marked as complete
		if ( $this->is_file_index_complete() ) {
			return 0;
		}

		// Resolve scan type from queue metadata (standard or deep).
		$queue = $this->get_cached_queue();
		$scan_type = ( $queue && isset( $queue['scan_type'] ) ) ? $queue['scan_type'] : 'standard';

		// Fetch User Exclusions
		if ( ! class_exists( 'SiteFort\\Modules\\Scanner\\Service\\ScannerSettings' ) ) {
			require_once dirname( __DIR__ ) . '/Service/ScannerSettings.php';
		}
		$settings = ScannerSettings::get_all();
		$user_exclusions = [];
		if ( ! empty( $settings['scanExclusions'] ) ) {
			$lines = explode( "\n", $settings['scanExclusions'] );
			foreach ( $lines as $line ) {
				$line = trim( $line );
				if ( ! empty( $line ) ) {
					$user_exclusions[] = $line;
				}
			}
		}

		$options = [
			'exclusions'  => $user_exclusions,
			'batch_limit' => $batch_limit,
		];

		$discovered = $this->collector->collect( $this->scan_id, $scan_type, $options );

		return $discovered;
	}

	/**
	 * Phase 1.5: System Checks.
	 * Run system checks (Reputation, Users, Content, Sensitive Data) sequentially.
	 * Returns true if ALL checks are complete, false if work remains.
	 *
	 * @return bool True if all checks are done.
	 */
	private function run_system_checks_phase(): bool {
		if ( ! $this->controller ) {
			$msg = 'System checks cannot run: scan controller is unavailable.';
			if ( class_exists( '\SiteFort\Modules\Scanner\Service\ScanLogger' ) ) {
				\SiteFort\Modules\Scanner\Service\ScanLogger::log( $msg, 3, 'error' );
			}
			ScanQueue::fail(
				$msg,
				[
					'error_code'        => 'controller_unavailable',
					'stage'             => 'system_checks',
					'retryable'         => true,
					'status_code'       => 500,
					'technical_message' => 'ScanOrchestrator missing ScanController in run_system_checks_phase',
				]
			);
			$this->reset_queue_cache();
			return false;
		}

		$queue = $this->get_cached_queue();
		if ( ! $queue ) {
			return false;
		}

		// Defined checks (DB-authoritative progression).
		$checks = [ 'reputation', 'users', 'content', 'sensitive_data', 'vulnerabilities' ];
		$component_status = ScanQueue::get_component_status( $queue, false );

		// Find the next non-terminal check.
		$next_slug = null;
		foreach ( $checks as $slug ) {
			$status = $component_status[ $slug ] ?? 'pending';
			if ( ! in_array( $status, [ 'complete', 'failed' ], true ) ) {
				$next_slug = $slug;
				break;
			}
		}

		if ( $next_slug === null ) {
			return true;
		}

		// Mark this check active before execution (with short grace to survive poll intervals).
		ScanQueue::set_component_status( $next_slug, 'active', 8 );

		try {
			$scan_result    = $this->controller->run_scan( $next_slug, $this->scan_id );
			$check_failed   = is_array( $scan_result ) && ! empty( $scan_result['failed'] );
			$check_complete = ! is_array( $scan_result ) || ! array_key_exists( 'complete', $scan_result )
				? true
				: (bool) $scan_result['complete'];

			if ( $check_failed ) {
				ScanQueue::set_component_status( $next_slug, 'failed' );
			} elseif ( $check_complete ) {
				ScanQueue::set_component_status( $next_slug, 'complete' );
			} else {
				ScanQueue::set_component_status( $next_slug, 'pending' );
			}
		} catch ( \Throwable $e ) {
			sitefort_log( "[SiteFort] System check failed ({$next_slug}): " . $e->getMessage() );
			ScanQueue::set_component_status( $next_slug, 'failed' );
		}
		$this->reset_queue_cache();

		return false;
	}

	/**
	 * Get scan start timestamp from DB or cache.
	 *
	 * @return int Unix timestamp (seconds).
	 */
	private function get_scan_start_time(): int {
		if ( $this->scan_start_time !== null ) {
			return $this->scan_start_time;
		}

		$cached = get_transient( 'sitefort_scan_start_' . $this->scan_id );
		if ( is_numeric( $cached ) && (int) $cached > 0 ) {
			$this->scan_start_time = (int) $cached;
			return $this->scan_start_time;
		}

		global $wpdb;
		$table = $this->table_identifier( $wpdb->prefix . 'sitefort_scan_queue' );
		$start_time = 0;
		try {
			// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,PluginCheck.Security.DirectDB.UnescapedDBParameter -- Uses a validated plugin-owned table identifier and a prepared job_id value.
			$start_time = (int) $wpdb->get_var(
				$wpdb->prepare(
					"SELECT created_at FROM {$table} WHERE job_id = %s",
					$this->scan_id
				)
			);
			// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,PluginCheck.Security.DirectDB.UnescapedDBParameter
		} catch ( \Throwable $e ) {
			$start_time = 0;
		}

		$start_time = absint( $start_time );
		if ( $start_time <= 0 ) {
			$start_time = time();
		}

		set_transient( 'sitefort_scan_start_' . $this->scan_id, $start_time, HOUR_IN_SECONDS );
		$this->scan_start_time = $start_time;
		return $this->scan_start_time;
	}

	/**
	 * Get scan max duration (seconds) from settings.
	 *
	 * @return int Max duration in seconds, or 0 if unlimited.
	 */
	private function get_scan_max_duration(): int {
		if ( $this->scan_max_duration !== null ) {
			return $this->scan_max_duration;
		}

		if ( ! class_exists( 'SiteFort\\Modules\\Scanner\\Service\\ScannerSettings' ) ) {
			require_once dirname( __DIR__ ) . '/Service/ScannerSettings.php';
		}

		$settings = ScannerSettings::get_all();
		$max = absint( $settings['scanMaxDuration'] ?? 0 );
		$this->scan_max_duration = $max > 0 ? $max : 0;
		return $this->scan_max_duration;
	}

	/**
	 * Get per-stage execution limit (seconds) from settings and PHP limit.
	 *
	 * @return int Stage execution limit in seconds.
	 */
	private function get_stage_execution_limit(): int {
		if ( $this->stage_execution_limit !== null ) {
			return $this->stage_execution_limit;
		}

		if ( ! class_exists( 'SiteFort\\Modules\\Scanner\\Service\\ScannerSettings' ) ) {
			require_once dirname( __DIR__ ) . '/Service/ScannerSettings.php';
		}

		$settings = ScannerSettings::get_all();
		$user_limit = absint( $settings['stageMaxExecutionTime'] ?? 0 );

		$ini_max = (int) ini_get( 'max_execution_time' );
		if ( $ini_max <= 0 ) {
			$ini_max = 30;
		}

		$limit = $ini_max;
		if ( $user_limit > 0 ) {
			$limit = min( $limit, $user_limit );
		}

		$this->stage_execution_limit = max( 1, $limit );
		return $this->stage_execution_limit;
	}

	/**
	 * Compute execution budget with safety margin.
	 *
	 * @return int Budgeted time in seconds.
	 */
	private function get_stage_execution_budget(): int {
		$limit         = $this->get_stage_execution_limit();
		$safety_margin = min( 5, (int) floor( $limit * 0.2 ) );
		return max( 1, $limit - $safety_margin );
	}

	/**
	 * Check if scan exceeded configured max duration.
	 *
	 * @return bool
	 */
	private function should_stop_for_scan_duration(): bool {
		$max = $this->get_scan_max_duration();
		if ( $max <= 0 ) {
			return false;
		}
		$start = $this->get_scan_start_time();
		return ( time() - $start ) >= $max;
	}

	/**
	 * Fail scan due to max duration exceeded.
	 *
	 * @return void
	 */
	private function fail_scan_due_to_timeout(): void {
		$msg = 'Scan time limit exceeded.';

		if ( ! class_exists( '\SiteFort\Modules\Scanner\Service\ScanLogger' ) ) {
			require_once __DIR__ . '/../Service/ScanLogger.php';
		}
		if ( class_exists( '\SiteFort\Modules\Scanner\Service\ScanLogger' ) ) {
			\SiteFort\Modules\Scanner\Service\ScanLogger::log( $msg, 3, 'error' );
		}

		if ( class_exists( '\SiteFort\Modules\Scanner\Repository\IssueRepository' ) ) {
			CloudResultProcessor::stage_skipped_issue( '(scan)', $msg, [ 'reason' => 'scan_timeout' ] );
		}

		if ( class_exists( '\SiteFort\Modules\Scanner\Queue\ScanQueue' ) ) {
			\SiteFort\Modules\Scanner\Queue\ScanQueue::fail(
				$msg,
				[
					'error_code'        => 'scan_timeout',
					'stage'             => 'scan_runtime',
					'retryable'         => true,
					'status_code'       => 408,
					'technical_message' => 'Scan max duration exceeded.',
				]
			);
		}
	}

	/**
	 * Phase 2: Hashing (AIMD Controlled).
	 * Applies unified pressure gating to prevent overload.
	 *
	 * @param array|null $pressure Pressure state.
	 * @param array|null $metrics Resource metrics for size-aware cost.
	 * @return int
	 */
	private function run_hashing_phase( ?array $pressure = null, ?array $metrics = null ): int {
		$window = $this->aimd->get_hash_window_for_phase( $pressure, $metrics );
		if ( $window <= 0 ) {
			return 0;
		}
		return $this->hash_service->process_hashing_batch( $this->scan_id, $window, $this->get_scan_type() );
	}

	/**
	 * Lazily resolve and cache the scan type for this orchestrator instance.
	 *
	 * Reads from the queue once; subsequent calls return the cached value.
	 *
	 * @return string Scan type constant (standard|deep).
	 */
	private function get_scan_type(): string {
		if ( $this->scan_type !== null ) {
			return $this->scan_type;
		}
		$queue            = $this->get_cached_queue();
		$this->scan_type  = ( $queue && isset( $queue['scan_type'] ) ) ? (string) $queue['scan_type'] : 'standard';
		return $this->scan_type;
	}

	/**
	 * Phase 3: Cloud Hash Check (AIMD Controlled).
	 * Applies unified pressure gating to avoid network amplification under load.
	 *
	 * @param array|null $pressure Pressure state.
	 * @param array|null $metrics Resource metrics for size-aware cost.
	 * @return int
	 */
	private function run_cloud_check_phase( ?array $pressure = null, ?array $metrics = null ): int {
		// Use a smaller cap than discovery to balance load
		$limit = min( $this->aimd->get_hash_window_for_phase( $pressure, $metrics ), self::MAX_HASH_CHECK_BATCH );
		if ( $limit <= 0 ) {
			return 0;
		}
		
		$files = $this->fsm->get_files_by_state( $this->scan_id, 'hash_complete', $limit );
        
        if ( count( $files ) < $limit ) {
            $stuck = $this->fsm->get_files_by_state( $this->scan_id, 'cloud_hash_check', $limit - count( $files ) );
            if ( ! empty( $stuck ) ) {
                $files = array_merge( $files, $stuck );
            }
        }

		if ( empty( $files ) ) {
			return 0;
		}
		
		// Buffering rule: flush when batch is full OR when at least 50 files are ready,
		// or when hashing has gone idle (final flush).
		$hashing_idle = $this->is_hash_production_idle();
		$file_index_complete = $this->is_file_index_complete();
		$min_flush = (int) max( 1, $limit / 5 ); // Flush at 20% of window, not 100%
		if ( count( $files ) < $min_flush && ( ! $hashing_idle || ! $file_index_complete ) ) {
			return 0;
		}

		$files = $this->refresh_stale_hash_snapshots( $files );
		if ( empty( $files ) ) {
			return 0;
		}

		$result = $this->cloud_service->check_hash( $files );

		if ( is_wp_error( $result ) ) {
			$error_data  = $result->get_error_data( 'api_error' );
			if ( empty( $error_data ) ) {
				$error_data = $result->get_error_data();
			}
			$status_code = is_array( $error_data ) ? (int) ( $error_data['status'] ?? 0 ) : 0;

			if ( $result->get_error_code() === 'rate_limited' || (int) $status_code === 429 ) {
				$this->defer_cloud_retry( $result );
				return 0;
			}

			$error_count = $this->handle_cloud_error( $result, 'hash_check' );
			if ( ! $this->is_current_scan_terminal() ) {
				$this->defer_cloud_retry( $result, $error_count );
			}
			return 0;
		}
		
		\SiteFort\Modules\Scanner\Queue\ScanQueue::reset_cloud_errors();
		$this->clear_cloud_retry_defer();
		
		$hash_results = $result['hash_results'] ?? [];
		$pinned_version = ScanQueue::get_pinned_signature_version( $this->scan_id );
		$signature_version = $pinned_version ?: ( $result['signature_version'] ?? 'unknown' );

		foreach ( $files as $file ) {
			$file_path = $file['file_path'];
			$verdict   = $hash_results[ $file_path ] ?? null;
			$file_hash_status = sanitize_key( (string) ( $file['hash_status'] ?? '' ) );
			$file_error_code  = sanitize_key( (string) ( $file['error_code'] ?? '' ) );
			$is_fake_image_candidate = $file_hash_status === IssueType::FAKE_IMAGE
				|| strpos( $file_hash_status, 'fake_image_' ) === 0
				|| $file_error_code === 'image_type_mismatch';

			if ( ! $verdict ) {
				$this->fsm->transition( (int) $file['id'], $file['state'], 'pending_upload' );
				continue;
			}

			$status = $verdict['status'] ?? 'unknown';
			
			// Persist hash verdict to guarantee repair eligibility survives deep-scan path.
			$verdict_payload = $verdict;
			if ( $file['state'] !== 'cloud_hash_check' ) {
				$this->fsm->transition(
					(int) $file['id'],
					'hash_complete',
					'cloud_hash_check',
					[
						'hash_status'           => $status,
						'scanned_with_signature'=> $signature_version,
						'scan_result'           => wp_json_encode( $verdict_payload ),
					]
				);
			} else {
				$this->file_repository->update(
					(int) $file['id'],
					[
						'hash_status'           => $status,
						'scanned_with_signature'=> $signature_version,
						'scan_result'           => wp_json_encode( $verdict_payload ),
					]
				);
			}

			switch ( $status ) {
				case 'verified':
					if ( $is_fake_image_candidate ) {
						$this->fsm->transition(
							(int) $file['id'],
							'cloud_hash_check',
							'pending_upload',
							[
								'hash_status'   => 'fake_image_verified_hash',
								'error_code'    => 'image_type_mismatch',
								'error_message' => 'Image validation failed locally; verified hash response bypassed for full cloud analysis',
							]
						);
						break;
					}
					$this->cloud_service->handle_verified_file( $file, $verdict );
					break;
				case 'modified':
					if ( $is_fake_image_candidate ) {
						$this->fsm->transition(
							(int) $file['id'],
							'cloud_hash_check',
							'pending_upload',
							[
								'hash_status'   => 'fake_image_modified_hash',
								'error_code'    => 'image_type_mismatch',
								'error_message' => 'Image validation failed locally; modified hash response bypassed for full cloud analysis',
							]
						);
						break;
					}
					$this->cloud_service->handle_modified_file( $file, $verdict );
					$this->fsm->transition( (int) $file['id'], 'cloud_hash_check', 'pending_upload', ['hash_status' => $status] );
					break;
				case 'unknown':
					if ( $is_fake_image_candidate ) {
						$this->fsm->transition(
							(int) $file['id'],
							'cloud_hash_check',
							'pending_upload',
							[
								'hash_status'   => 'fake_image_unknown_hash',
								'error_code'    => 'image_type_mismatch',
								'error_message' => 'Image validation failed locally; unknown hash response bypassed for full cloud analysis',
							]
						);
						break;
					}
					$this->cloud_service->handle_unknown_file( $file, $verdict );
					$this->fsm->transition( (int) $file['id'], 'cloud_hash_check', 'pending_upload', ['hash_status' => $status] );
					break;
				case 'safe':
					if ( $is_fake_image_candidate ) {
						$this->fsm->transition(
							(int) $file['id'],
							'cloud_hash_check',
							'pending_upload',
							[
								'hash_status'   => 'fake_image_safe_hash',
								'error_code'    => 'image_type_mismatch',
								'error_message' => 'Image validation failed locally; safe hash response bypassed for full cloud analysis',
							]
						);
						break;
					}
					if ( empty( $signature_version ) || $signature_version === 'unknown' ) {
						$this->fsm->transition( (int) $file['id'], 'cloud_hash_check', 'pending_upload', [ 'hash_status' => 'safe_stale' ] );
						break;
					}
					if ( ! empty( $verdict['signature_version'] ) && $verdict['signature_version'] !== $signature_version ) {
						$this->fsm->transition( (int) $file['id'], 'cloud_hash_check', 'pending_upload', [ 'hash_status' => 'safe_stale' ] );
						break;
					}
					$this->cloud_service->handle_safe_file( $file, $verdict, $signature_version );
					break;
				case 'malicious':
					$this->cloud_service->handle_malicious_file( $file, $verdict );
					break;
				default:
					$this->fsm->transition( (int) $file['id'], 'cloud_hash_check', 'pending_upload', ['hash_status' => $status] );
					break;
			}
		}

		// Accumulate unindexed WP core versions seen during this batch.
		$unindexed = $result['unindexed_wp_versions'] ?? [];
		if ( ! empty( $unindexed ) ) {
			$tkey     = 'sitefort_unindexed_wpv_' . $this->scan_id;
			$existing = (array) ( get_transient( $tkey ) ?: [] );
			$merged   = CloudScanService::normalize_unindexed_wp_versions( array_merge( $existing, $unindexed ) );
			set_transient( $tkey, $merged, 7 * DAY_IN_SECONDS );
		}

		// Credit stop: mark all remaining pre-upload files as skipped for this scan.
		if ( ! empty( $result['credit_limit_reached'] ) ) {
			$this->stop_scan_for_credit_exhaustion(
				[
					'credits_remaining' => $result['credits_remaining'] ?? null,
					'credits_total'     => $result['credits_total'] ?? null,
					'credit_reset_at'   => $result['credit_reset_at'] ?? null,
				]
			);
		}

		return count( $files );
	}

	/**
	 * Rehash files whose component or mtime snapshot drifted before cloud lookup.
	 *
	 * @param array<int,array<string,mixed>> $files
	 * @return array<int,array<string,mixed>>
	 */
	private function refresh_stale_hash_snapshots( array $files ): array {
		$ready = [];

		foreach ( $files as $file ) {
			$path  = (string) ( $file['file_path'] ?? '' );
			$state = (string) ( $file['state'] ?? 'hash_complete' );
			$id    = (int) ( $file['id'] ?? 0 );
			if ( $id <= 0 || $path === '' ) {
				continue;
			}

			$absolute_path = ABSPATH . $path;
			if ( ! is_file( $absolute_path ) || ! is_readable( $absolute_path ) ) {
				$this->fsm->transition(
					$id,
					$state,
					'error',
					[
						'error_code'    => 'file_missing',
						'error_message' => 'File missing or unreadable before cloud hash check',
					]
				);
				continue;
			}

			if ( $this->collector->file_state_snapshot_matches_current( $file ) ) {
				$ready[] = $file;
				continue;
			}

			$refreshed = $this->collector->build_hash_snapshot( $path, $absolute_path );
			if ( empty( $refreshed['stable'] ) || false === ( $refreshed['hash'] ?? false ) ) {
				$this->fsm->transition(
					$id,
					$state,
					'scan_skipped',
					[
						'skip_reason'   => 'changed_during_scan',
						'error_code'    => 'changed_during_scan',
						'error_message' => 'File changed during scan before cloud hash check; rerun after updates finish',
					]
				);
				$this->log_scan_warning( '[Skipped] ' . $path . ' - file changed during scan before cloud hash check' );
				continue;
			}

			$update = [
				'file_hash'              => (string) $refreshed['hash'],
				'file_size'              => (int) $refreshed['size'],
				'file_type'              => (string) ( $refreshed['info']['type'] ?? ( $file['file_type'] ?? 'other' ) ),
				'component_slug'         => (string) ( $refreshed['info']['slug'] ?? '' ),
				'component_version'      => (string) ( $refreshed['info']['version'] ?? '' ),
				'component_locale'       => (string) ( $refreshed['info']['locale'] ?? '' ),
				'file_mtime_at_hash'     => (int) $refreshed['mtime'],
				'hash_status'            => 'unknown',
				'scan_result'            => null,
				'scanned_with_signature' => null,
				'cloud_job_id'           => null,
				'error_code'             => 'snapshot_drift_rehashed',
				'error_message'          => 'File/component snapshot changed before cloud hash check; rehashed with current metadata',
			];

			$result = $this->fsm->transition( $id, $state, 'hash_complete', $update );
			if ( is_wp_error( $result ) ) {
				$this->log_scan_warning( '[SiteFort] Failed to refresh stale hash snapshot for ' . $path . ': ' . $result->get_error_message() );
				continue;
			}

			$ready[] = array_merge( $file, $update, [ 'state' => 'hash_complete' ] );
		}

		return $ready;
	}

	private function log_scan_warning( string $message ): void {
		if ( class_exists( '\SiteFort\Modules\Scanner\Service\ScanLogger' ) ) {
			\SiteFort\Modules\Scanner\Service\ScanLogger::log( $message, 2, 'warning' );
		}
	}

	/**
	 * Skip all remaining non-terminal file states when credits are exhausted.
	 *
	 * Transitions the remaining file pipeline states to scan_skipped with
	 * skip_reason='credit_exhausted'. Sets a transient so subsequent ticks
	 * bail out immediately without touching the DB.
	 */
	private function skip_remaining_files_credit_exhausted(): int {
		$skippable_states = [ 'discovered', 'hashing', 'hash_complete', 'cloud_hash_check', 'pending_upload', 'uploading', 'uploaded', 'scan_requested', 'scanning' ];
		$skipped_count    = 0;

		foreach ( $skippable_states as $state ) {
			// Process in batches of 500 to avoid large IN queries.
			do {
				$files = $this->fsm->get_files_by_state( $this->scan_id, $state, 500 );
				foreach ( $files as $file ) {
					$this->fsm->transition(
						(int) $file['id'],
						$state,
						'scan_skipped',
						[
							'skip_reason'   => 'credit_exhausted',
							'error_code'    => 'credits_exhausted',
							'error_message' => 'Cloud scan credits exhausted; file skipped for this scan',
							'cloud_job_id'  => null,
						]
					);
					$skipped_count++;
				}
			} while ( count( $files ) === 500 );
		}

		// Signal future ticks to skip hash work immediately.
		set_transient( 'sitefort_credit_exhausted_' . $this->scan_id, true, HOUR_IN_SECONDS );

		if ( class_exists( 'SiteFort\Modules\Scanner\Service\ScanLogger' ) ) {
			\SiteFort\Modules\Scanner\Service\ScanLogger::log(
				sprintf( 'Cloud scan credit limit reached. %d files marked as skipped.', $skipped_count ),
				2,
				'warning'
			);
		}

		return $skipped_count;
	}

	/**
	 * Persist the latest cloud credit fields for the scanner UI.
	 *
	 * @param array<string,mixed> $credit_context
	 * @return void
	 */
	private function persist_credit_exhaustion_fields( array $credit_context = [] ): void {
		if ( ! class_exists( '\SiteFort\Modules\Cloud\Service\CloudSettings' ) ) {
			require_once dirname( dirname( __DIR__ ) ) . '/Cloud/Service/CloudSettings.php';
		}

		if ( ! class_exists( '\SiteFort\Modules\Cloud\Service\CloudSettings' ) ) {
			return;
		}

		$credit_update = [ 'scan_credit_limit_reached' => true ];

		if ( array_key_exists( 'credits_remaining', $credit_context ) && $credit_context['credits_remaining'] !== null ) {
			$credit_update['scan_credits_remaining'] = (int) $credit_context['credits_remaining'];
		}
		if ( array_key_exists( 'credits_total', $credit_context ) && $credit_context['credits_total'] !== null ) {
			$credit_update['scan_credits_total'] = (int) $credit_context['credits_total'];
		}
		if ( array_key_exists( 'credit_reset_at', $credit_context ) && is_string( $credit_context['credit_reset_at'] ) ) {
			$credit_update['scan_credit_reset_at'] = $credit_context['credit_reset_at'];
		}

		\SiteFort\Modules\Cloud\Service\CloudSettings::update_stored_fields( $credit_update );
	}

	/**
	 * Mark locally tracked cloud jobs as failed after a terminal credit stop.
	 *
	 * @return void
	 */
	private function fail_active_cloud_jobs_for_credit_exhaustion(): void {
		$jobs = $this->job_repository->get_by_scan_id( $this->scan_id );

		foreach ( $jobs as $job ) {
			$status = (string) ( $job['status'] ?? '' );
			$job_id = (string) ( $job['job_id'] ?? '' );
			if ( '' === $job_id || ! in_array( $status, [ 'queued', 'submitted', 'processing' ], true ) ) {
				continue;
			}

			$this->job_repository->update(
				$job_id,
				[
					'status'        => 'failed',
					'error_message' => 'Cloud scan credits exhausted; scan stopped before results completed',
					'completed_at'  => time(),
				]
			);
		}
	}

	/**
	 * Stop the scan immediately when the cloud credit window is exhausted.
	 *
	 * @param array<string,mixed> $credit_context
	 * @return void
	 */
	private function stop_scan_for_credit_exhaustion( array $credit_context = [] ): void {
		$queue = $this->get_cached_queue();
		if ( ! is_array( $queue ) || ! in_array( (string) ( $queue['status'] ?? '' ), [ 'pending', 'running' ], true ) ) {
			return;
		}

		$this->persist_credit_exhaustion_fields( $credit_context );
		$this->skip_remaining_files_credit_exhausted();
		$this->fail_active_cloud_jobs_for_credit_exhaustion();
		delete_transient( 'sitefort_cloud_job_' . $this->scan_id );

		ScanQueue::fail(
			self::CREDIT_EXHAUSTED_STOP_MESSAGE,
			[
				'error_code'        => 'credits_exhausted',
				'stage'             => 'cloud_credit_limit',
				'retryable'         => false,
				'status_code'       => 402,
				'technical_message' => 'Cloud credit window exhausted during active scan',
			]
		);
		$this->reset_queue_cache();
	}

	/**
	 * Phase 4: Upload (AIMD Controlled).
	 */
	private function run_upload_phase( array $preselected = [] ): int {
		$hashing_complete = $this->check_hashing_complete();
		$files = ! empty( $preselected ) ? $preselected : $this->select_upload_candidates( $hashing_complete );

		if ( empty( $files ) ) {
			return 0;
		}

		// Filter out missing/unreadable files to avoid retrying the same item forever
		$valid_files = [];
		foreach ( $files as $file ) {
			$path = (string) ( $file['file_path'] ?? '' );
			$full_path = ABSPATH . $path;
			if ( CloudResultProcessor::is_runtime_artifact_path( $path ) ) {
				$this->fsm->transition(
					(int) $file['id'],
					'pending_upload',
					'scan_skipped',
					[
						'skip_reason'   => 'runtime_artifact',
						'error_code'    => 'runtime_artifact_ignored',
						'error_message' => 'SiteFort runtime data file changed during scan and was ignored',
					]
				);
				CloudResultProcessor::stage_skipped_issue(
					$path,
					'SiteFort runtime data file changed during scan and was ignored',
					[ 'error_code' => 'runtime_artifact_ignored' ]
				);
				continue;
			}

			if ( empty( $path ) || ! file_exists( $full_path ) || ! is_readable( $full_path ) ) {
				$this->fsm->transition(
					(int) $file['id'],
					'pending_upload',
					'error',
					[
						'error_code'    => 'file_missing',
						'error_message' => 'File missing or unreadable during upload',
					]
				);
				CloudResultProcessor::stage_skipped_issue(
					$path ?: '(unknown)',
					'Upload skipped: file missing or unreadable during upload',
					[ 'error_code' => 'file_missing' ]
				);
				continue;
			}

			// Re-verify both file size and file hash immediately before upload so
			// same-size mutations cannot be uploaded under a stale hash label.
			$current_snapshot = $this->collector->build_hash_snapshot( $path, $full_path );
			if ( empty( $current_snapshot['stable'] ) || false === ( $current_snapshot['hash'] ?? false ) ) {
				$this->fsm->transition(
					(int) $file['id'],
					'pending_upload',
					'scan_skipped',
					[
						'skip_reason'   => 'changed_during_scan',
						'error_code'    => 'changed_during_scan',
						'error_message' => 'File changed while verifying upload snapshot; rerun after updates finish',
					]
				);
				CloudResultProcessor::stage_skipped_issue(
					$path,
					'Upload skipped: file changed while verifying upload snapshot',
					[ 'error_code' => 'changed_during_scan' ]
				);
				continue;
			}

			$current_size = (int) $current_snapshot['size'];
			$current_hash = (string) $current_snapshot['hash'];

			$expected_size = (int) ( $file['file_size'] ?? 0 );
			$expected_hash = (string) ( $file['file_hash'] ?? '' );
			if (
				$expected_size < 0 ||
				$expected_hash === '' ||
				$current_size !== $expected_size ||
				! hash_equals( $expected_hash, $current_hash )
			) {
				// File changed during the scan. Re-queue back to hash_complete with updated hash.
				$this->fsm->transition(
					(int) $file['id'],
					'pending_upload',
					'hash_complete',
					[
						'file_hash'              => $current_hash,
						'file_size'              => $current_size,
						'file_type'              => (string) ( $current_snapshot['info']['type'] ?? ( $file['file_type'] ?? 'other' ) ),
						'component_slug'         => (string) ( $current_snapshot['info']['slug'] ?? '' ),
						'component_version'      => (string) ( $current_snapshot['info']['version'] ?? '' ),
						'component_locale'       => (string) ( $current_snapshot['info']['locale'] ?? '' ),
						'file_mtime_at_hash'     => (int) $current_snapshot['mtime'],
						'hash_status'            => 'unknown',
						'scan_result'            => null,
						'scanned_with_signature' => null,
						'cloud_job_id'           => null,
						'error_code'             => 'changed_during_scan',
						'error_message'          => 'File content changed between hashing and upload; re-queued for re-check',
					]
				);
				continue;
			}

			$valid_files[] = $file;
		}

		if ( empty( $valid_files ) ) {
			return 0;
		}

		// Mark as uploading + increment attempt so a "poison pill" file can't starve the scan forever.
		foreach ( $valid_files as $file ) {
			$this->fsm->transition(
				(int) $file['id'],
				'pending_upload',
				'uploading',
				[
					'upload_attempt' => ( (int) ( $file['upload_attempt'] ?? 0 ) ) + 1,
				]
			);
		}

		// Upload Trigger Logic is now handled by the Execution Gate Loop (Caller)
		// which checks buffer size/failures. 
		// If we are here, we MUST upload what we grabbed (up to window size).

		$result = $this->cloud_service->get_secure_link( $valid_files );

		if ( is_wp_error( $result ) ) {
			$error_data = $result->get_error_data();
			$status_code = $error_data['status'] ?? 0;
			$api_detail = is_array( $error_data['detail'] ?? null ) ? $error_data['detail'] : [];
			$api_error_code = sanitize_key( (string) ( $api_detail['error_code'] ?? '' ) );

			$is_429 = ( 429 === $status_code || 429 === $result->get_error_code() || false !== strpos( $result->get_error_message(), '429' ) );
			$this->aimd->record_upload_failure( $is_429 );

			sitefort_log( '[SiteFort] Upload Failed: ' . $result->get_error_message() . ( $is_429 ? ' (Rate Limit - Backing off)' : '' ) );

			if ( ! class_exists( '\SiteFort\Modules\Scanner\Service\ScanLogger' ) ) {
				require_once __DIR__ . '/../Service/ScanLogger.php';
			}
			if ( class_exists( '\SiteFort\Modules\Scanner\Service\ScanLogger' ) ) {
				\SiteFort\Modules\Scanner\Service\ScanLogger::log(
					'Upload failed: ' . $result->get_error_message(),
					$is_429 ? 2 : 3,
					$is_429 ? 'warning' : 'error'
				);
			}

			// Credits exhausted during upload-link creation (race between preflight and upload).
			// Normalize to the same skip behavior used in check-hash credit exhaustion.
			if ( $api_error_code === 'credits_exhausted' ) {
				$this->stop_scan_for_credit_exhaustion(
					[
						'credits_remaining' => $api_detail['credits_remaining'] ?? null,
						'credits_total'     => $api_detail['credits_total'] ?? null,
						'credit_reset_at'   => $api_detail['credit_reset_at'] ?? null,
					]
				);
				return 0;
			}

				// Recover strict hash-registration desync by sending files back through
				// hash-check before the next upload request.
				if ( $api_error_code === 'hash_not_registered' ) {
					foreach ( $valid_files as $file ) {
						$transition = $this->fsm->transition(
							(int) $file['id'],
							'uploading',
							'hash_complete',
							[
								'file_hash'              => (string) ( $file['file_hash'] ?? '' ),
								'file_size'              => (int) ( $file['file_size'] ?? 0 ),
								'hash_status'            => 'unknown',
								'scanned_with_signature' => null,
								'cloud_job_id'           => null,
								'error_code'             => 'hash_registry_desync',
								'error_message'          => 'Cloud hash registration missing; re-checking hash before upload retry',
							]
						);

						if ( is_wp_error( $transition ) ) {
							$this->fsm->transition( (int) $file['id'], 'uploading', 'pending_upload' );
						}
					}
					return 0;
				}

				// Roll back files to pending_upload (they were not confirmed uploaded).
				foreach ( $valid_files as $file ) {
					$this->fsm->transition( (int) $file['id'], 'uploading', 'pending_upload' );
				}
				if ( ! $is_429 ) {
					$this->handle_cloud_error( $result, 'upload' );
			}
			return 0;
		}

		// Success
		$this->aimd->record_upload_success();
		\SiteFort\Modules\Scanner\Queue\ScanQueue::reset_cloud_errors();
		
		$job_id = $result['job_id'];
		$uploaded_hashes = $result['uploaded_hashes'] ?? [];
		
		$this->current_job_id = $job_id;
		set_transient( 'sitefort_cloud_job_' . $this->scan_id, $job_id, HOUR_IN_SECONDS );

		// OPTIMIZATION: If no files were actually uploaded (e.g. S3 failure), kill the job immediately.
		// Prevents polling "zombie" jobs that will never complete.
		if ( empty( $uploaded_hashes ) ) {
			$this->job_repository->update( $job_id, [ 'status' => 'failed', 'error_message' => 'upload_failed_locally' ] );
		}

		$uploaded_count = 0;
		foreach ( $valid_files as $file ) {
			$file_hash = $file['file_hash'] ?? '';
			if ( in_array( $file_hash, $uploaded_hashes, true ) ) {
				$this->fsm->transition( (int) $file['id'], 'uploading', 'uploaded', [ 'cloud_job_id' => $job_id ] );
				$uploaded_count++;
			} else {
				// Not included in uploaded_hashes -> requeue this file.
				$this->fsm->transition( (int) $file['id'], 'uploading', 'pending_upload' );
			}
		}

		return $uploaded_count;
	}

	/**
	 * Select upload candidates to respect 2MB/50 batching rules.
	 *
	 * When hashing is complete, smaller final batches are allowed.
	 *
	 * @param bool $hashing_complete
	 * @return array
	 */
	private function resolve_upload_max_size_bytes( array $payload ): int {
		$size_mb = (int) ( $payload['max_size_mb'] ?? 0 );
		if ( $size_mb <= 0 ) {
			$size_mb = 8;
		}

		return $size_mb * 1024 * 1024;
	}

	private function has_cloud_upload_capacity(): bool {
		return $this->get_active_cloud_job_count() < $this->get_max_active_cloud_jobs();
	}

	private function get_active_cloud_job_count(): int {
		return $this->job_repository->count_by_scan_and_statuses(
			$this->scan_id,
			[ 'submitted', 'processing' ]
		);
	}

	private function get_max_active_cloud_jobs(): int {
		$limit = self::MAX_ACTIVE_CLOUD_JOBS;
		if ( function_exists( 'apply_filters' ) ) {
			$limit = (int) apply_filters( 'sitefort_scanner_max_active_cloud_jobs', $limit, $this->scan_id );
		}

		return max( 1, min( 50, $limit ) );
	}

	/**
	 * Select upload candidates to respect batching rules.
	 *
	 * @param bool $hashing_complete
	 * @param int  $max_files_override
	 * @param int  $max_size_override
	 * @return array
	 */
	private function select_upload_candidates(
		bool $hashing_complete,
		int $max_files_override = 0,
		int $max_size_override = 0
	): array {
		$max_files = $max_files_override > 0 ? min( CloudScanService::MAX_FILES_PER_BATCH, $max_files_override ) : CloudScanService::MAX_FILES_PER_BATCH;
		$max_size  = $max_size_override > 0 ? $max_size_override : 2 * 1024 * 1024;

		// Pull more than we need to build a proper batch.
		$candidates = $this->file_repository->get_pending_upload_candidates( $this->scan_id, 200 );
		if ( empty( $candidates ) ) {
			return [];
		}

		$selected = [];
		$total_size = 0;

		foreach ( $candidates as $file ) {
			// Transition files that have exhausted upload attempts to error state
			// so check_completion() can see them as terminal.
			if ( (int) ( $file['upload_attempt'] ?? 0 ) >= 3 ) {
				$this->fsm->transition(
					(int) $file['id'],
					'pending_upload',
					'error',
					[
						'error_code'    => 'upload_cycle_exhausted',
						'error_message' => 'Cloud upload failed after multiple attempts',
					]
				);
				CloudResultProcessor::stage_skipped_issue(
					(string) ( $file['file_path'] ?? '' ),
					'Upload failed: cloud rejected file after multiple attempts',
					[ 'error_code' => 'upload_cycle_exhausted', 'upload_attempts' => (int) ( $file['upload_attempt'] ?? 0 ) ]
				);
				continue;
			}

			$size = (int) ( $file['file_size'] ?? 0 );

			// Always allow a >2MB file to be its own batch.
			if ( $size > $max_size && empty( $selected ) ) {
				$selected[] = $file;
				$total_size += $size;
				break;
			}

			$selected[] = $file;
			$total_size += $size;

			if ( count( $selected ) >= $max_files || $total_size >= $max_size ) {
				break;
			}
		}

		// During hashing, flush only full batches or aged partial batches.
		if ( ! $hashing_complete ) {
			if ( ! $this->should_flush_partial_upload_batch( $selected, $max_files, $total_size, $max_size ) ) {
				return [];
			}
		}

		return $selected;
	}

	private function should_flush_partial_upload_batch( array $selected, int $max_files, int $total_size, int $max_size ): bool {
		if ( empty( $selected ) ) {
			return false;
		}

		if ( count( $selected ) >= $max_files || $total_size >= $max_size ) {
			return true;
		}

		$oldest_ready_at = PHP_INT_MAX;
		foreach ( $selected as $file ) {
			$ready_at = (int) ( $file['updated_at'] ?? 0 );
			if ( $ready_at <= 0 ) {
				$ready_at = (int) ( $file['created_at'] ?? 0 );
			}
			if ( $ready_at > 0 ) {
				$oldest_ready_at = min( $oldest_ready_at, $ready_at );
			}
		}

		return $oldest_ready_at !== PHP_INT_MAX
			&& ( time() - $oldest_ready_at ) >= self::PARTIAL_UPLOAD_FLUSH_AFTER;
	}

	/**
	 * Helper to check if hashing is done.
	 */
	private function check_hashing_complete(): bool {
		if ( $this->hashing_complete_cache === true ) {
			return true;
		}
		$now = time();
		if ( $this->hashing_complete_cache === false && ( $now - $this->hashing_complete_cache_time ) < 3 ) {
			return false;
		}
		$result = ! $this->fsm->has_files_in_states(
			$this->scan_id,
			[ 'discovered', 'hashing', 'hash_complete', 'cloud_hash_check' ]
		);
		$this->hashing_complete_cache      = $result;
		$this->hashing_complete_cache_time = $now;
		return $result;
	}

	/**
	 * Check if hash production is idle (no discovered or hashing files).
	 * This allows the cloud hash phase to flush a partial batch when no more hashes are coming.
	 */
	private function is_hash_production_idle(): bool {
		// Single query with IN clause instead of 2 separate SELECT queries.
		return ! $this->fsm->has_files_in_states( $this->scan_id, [ 'discovered', 'hashing' ] );
	}

	public function get_verified_count(): int {
		$progress = $this->fsm->get_scan_progress( $this->scan_id );
		return (
			( $progress['completed'] ?? 0 ) +
			( $progress['flagged'] ?? 0 ) +
			( $progress['error'] ?? 0 ) +
			( $progress['verified'] ?? 0 ) +
			( $progress['safe'] ?? 0 ) +
			( $progress['scan_skipped'] ?? 0 )
		);
	}

	/**
	 * Phase 6: Check Completion.
	 * Determine if the scan is complete (all files in terminal states).
	 *
	 * @return bool True if scan is complete.
	 */
	private function check_completion(): bool {
		// CRITICAL: Ensure file indexing is actually finished before declaring scan complete
		if ( ! $this->is_file_index_complete() ) {
			return false;
		}

		// All system checks must reach a terminal state before finalizing the scan;
		// without this gate, the queue can flip to complete while late checks are still
		// pending/active, causing the UI to jump out-of-sequence.
		if ( ! $this->are_system_checks_complete() ) {
			return false;
		}

		// Scan is complete only when every indexed/discovered file is terminal.
		if ( ! $this->is_file_pipeline_complete() ) {
			return false;
		}

		// NOTE: Do NOT commit here. ScanQueue::complete() → publish() handles the
		// atomic swap (hard-clean → commit → swap). Committing here would insert
		// issues that publish()'s hard-clean then deletes, relying on a second commit
		// to re-insert them - a fragile pattern that loses issues on silent SQL errors.
		return true;
	}

	/**
	 * True when every file in the FSM has reached a terminal state.
	 * This means discovery, hashing, uploading, and cloud polling are all done.
	 * Does NOT require system checks (reputation, vulns, etc.) to be complete.
	 */
	private function is_file_pipeline_complete(): bool {
		$progress = $this->fsm->get_scan_progress( $this->scan_id );

		$terminal_states = [
			'completed',
			'flagged',
			'error',
			'verified',
			'safe',
			'scan_skipped',
		];

		$terminal_count = 0;
		foreach ( $terminal_states as $state ) {
			$terminal_count += $progress[ $state ] ?? 0;
		}

		$fsm_total        = (int) ( $progress['total'] ?? 0 );
		$file_index_total = $this->file_index_repository
			? (int) $this->file_index_repository->get_total_files( $this->scan_id )
			: 0;
		$expected_total   = max( $fsm_total, $file_index_total );

		return $expected_total > 0 && $terminal_count >= $expected_total;
	}

	/**
	 * Check whether all system checks are complete (or failed terminally).
	 *
	 * Uses durable queue component_status as source of truth.
	 *
	 * @return bool
	 */
	private function are_system_checks_complete(): bool {
		$checks = [ 'reputation', 'users', 'content', 'sensitive_data', 'vulnerabilities' ];

		$queue = $this->get_cached_queue();
		if ( ! $queue ) {
			return false;
		}

		$component_status = ScanQueue::get_component_status( $queue, false );
		foreach ( $checks as $slug ) {
			$status = $component_status[ $slug ] ?? 'pending';
			if ( ! in_array( $status, [ 'complete', 'failed' ], true ) ) {
				return false;
			}
		}

		return true;
	}

	/**
	 * Resolve file index completion across transient and DB state.
	 *
	 * @return bool
	 */
	private function is_file_index_complete(): bool {
		if ( $this->file_index_complete_cache === true ) {
			return true;
		}

		$transient_key = 'sitefort_file_index_complete_' . $this->scan_id;
		if ( get_transient( $transient_key ) ) {
			$this->file_index_complete_cache = true;
			return true;
		}

		$legacy_key = 'sitefort_discovery_complete_' . $this->scan_id;
		if ( get_transient( $legacy_key ) ) {
			set_transient( $transient_key, true, HOUR_IN_SECONDS );
			$this->file_index_complete_cache = true;
			return true;
		}

		if ( $this->file_index_repository && $this->file_index_repository->is_complete( $this->scan_id ) ) {
			set_transient( $transient_key, true, HOUR_IN_SECONDS );
			$this->file_index_complete_cache = true;
			return true;
		}

		return false;
	}

	/**
	 * Get scan progress for reporting.
	 *
	 * @return array Progress statistics.
	 */
	public function get_progress(): array {
		return $this->fsm->get_scan_progress( $this->scan_id );
	}

	/**
	 * Get cloud job status for monitoring.
	 *
	 * @return array Cloud job statistics.
	 */
	public function get_cloud_stats(): array {
		$jobs = $this->job_repository->get_by_scan_id( $this->scan_id );

		$stats = [
			'total_jobs'  => count( $jobs ),
			'submitted'   => 0,
			'processing'  => 0,
			'completed'   => 0,
			'failed'      => 0,
			'timeout'     => 0,
		];

		foreach ( $jobs as $job ) {
			$status = $job['status'] ?? 'queued';
			if ( isset( $stats[ $status ] ) ) {
				$stats[ $status ]++;
			}
		}

		return $stats;
	}

	/**
	 * Poll for scan results from Intel Engine (PRIMARY delivery method).
	 *
	 * Retrieves and processes scan results from cloud jobs.
	 * Uses **Vectorized Polling** to check all pending jobs in a single request.
	 *
	 * @return int Number of jobs processed.
	 */
	public function poll_results(): int {
		$jobs = $this->job_repository->get_jobs_needing_poll( $this->scan_id );
		if ( empty( $jobs ) ) {
			return 0;
		}

		$result = $this->process_polled_jobs( $jobs );
		if ( is_wp_error( $result ) ) {
			return 0;
		}

		return (int) ( $result['processed'] ?? 0 );
	}

	/**
	 * Poll and process a batch of known cloud job rows.
	 *
	 * @param array<int,array<string,mixed>> $jobs
	 * @return array<string,mixed>|\WP_Error
	 */
	private function process_polled_jobs( array $jobs ) {
		if ( empty( $jobs ) ) {
			return [
				'processed'             => 0,
				'pending_job_ids'       => [],
				'completed_job_ids'     => [],
				'failed_job_ids'        => [],
				'requeued_upload'       => false,
				'youngest_pending_age' => null,
			];
		}

		$job_ids_to_poll = [];
		$jobs_map = [];
		$now = time();

		foreach ( $jobs as $job ) {
			$job_id = (string) ( $job['job_id'] ?? '' );
			if ( '' === $job_id ) {
				continue;
			}

			$job_ids_to_poll[] = $job_id;
			$jobs_map[ $job_id ] = $job;
		}

		if ( empty( $job_ids_to_poll ) ) {
			return [
				'processed'             => 0,
				'pending_job_ids'       => [],
				'completed_job_ids'     => [],
				'failed_job_ids'        => [],
				'requeued_upload'       => false,
				'youngest_pending_age' => null,
			];
		}

		$batch_results = $this->cloud_service->poll_batch_results( $job_ids_to_poll );
		if ( is_wp_error( $batch_results ) ) {
			$this->handle_cloud_error( $batch_results, 'poll_results' );
			return $batch_results;
		}

		\SiteFort\Modules\Scanner\Queue\ScanQueue::reset_cloud_errors();
		foreach ( $job_ids_to_poll as $job_id ) {
			$this->job_repository->update( $job_id, [ 'last_poll_at' => $now ] );
		}

		$summary = [
			'processed'             => 0,
			'pending_job_ids'       => [],
			'completed_job_ids'     => [],
			'failed_job_ids'        => [],
			'requeued_upload'       => false,
			'youngest_pending_age' => null,
		];

		foreach ( $job_ids_to_poll as $job_id ) {
			$result_data = $batch_results[ $job_id ] ?? null;
			$job = $jobs_map[ $job_id ];
			$status = is_array( $result_data ) ? (string) ( $result_data['status'] ?? 'pending' ) : 'missing';

			if ( 'completed' === $status ) {
				$this->result_processor->process_scan_results( $job_id, $result_data['results'] ?? [] );
				$this->job_repository->update(
					$job_id,
					[
						'status'       => 'completed',
						'completed_at' => time(),
					]
				);
				$summary['processed']++;
				$summary['completed_job_ids'][] = $job_id;
				continue;
			}

			if ( 'failed' === $status ) {
				$this->job_repository->update(
					$job_id,
					[
						'status'        => 'failed',
						'error_message' => $result_data['error'] ?? 'Server reported failure',
						'completed_at'  => time(),
					]
				);
				$this->result_processor->mark_job_files_as_error( $job_id, 'server_failure', $result_data['error'] ?? 'Analysis failed' );
				$summary['failed_job_ids'][] = $job_id;
				continue;
			}

			if ( in_array( $status, [ 'awaiting_upload', 'uploading', 'not_found', 'missing' ], true ) ) {
				$submitted_at = (int) ( $job['submitted_at'] ?? 0 );
				$age = $submitted_at > 0 ? ( time() - $submitted_at ) : 0;

				if ( $submitted_at > 0 && $age >= self::AWAITING_UPLOAD_REQUEUE_AFTER ) {
					$retry_count = (int) ( $job['retry_count'] ?? 0 );
					$max_retries = (int) ( $job['max_retries'] ?? 3 );
					$state_desc = in_array( $status, [ 'not_found', 'missing' ], true )
						? 'not returned by cloud poll'
						: $status;

					if ( $retry_count < $max_retries ) {
						$this->job_repository->update(
							$job_id,
							[
								'status'        => 'failed',
								'error_message' => sprintf( 'Job %s for %ds; re-queueing upload batch', $state_desc, $age ),
								'completed_at'  => time(),
								'retry_count'   => $retry_count + 1,
							]
						);
						$this->requeue_job_files_for_upload( $job_id );
						$this->aimd->record_upload_failure( false );
						$summary['requeued_upload'] = true;
						$summary['failed_job_ids'][] = $job_id;
					} else {
						$this->job_repository->update(
							$job_id,
							[
								'status'        => 'failed',
								'error_message' => sprintf( 'Job %s; retries exhausted (%d)', $state_desc, $max_retries ),
								'completed_at'  => time(),
							]
						);
						$this->result_processor->mark_job_files_as_error( $job_id, 'awaiting_upload_timeout', 'Cloud did not confirm job status; retries exhausted' );
						$summary['failed_job_ids'][] = $job_id;
					}
				} else {
					$this->record_pending_job( $summary, $job_id, $job, $now );
				}
				continue;
			}

			if ( $this->is_job_timed_out( $job ) ) {
				$retry_count = (int) ( $job['retry_count'] ?? 0 );
				$max_retries = (int) ( $job['max_retries'] ?? 3 );

				if ( $retry_count < $max_retries ) {
					$this->job_repository->update(
						$job_id,
						[
							'status'       => 'submitted',
							'retry_count'  => $retry_count + 1,
							'timeout_at'   => time() + 900,
							'last_poll_at' => null,
						]
					);
					$this->record_pending_job( $summary, $job_id, $job, $now );
				} else {
					$this->job_repository->update(
						$job_id,
						[
							'status'        => 'failed',
							'error_message' => sprintf( 'Job failed after %d retries', $max_retries ),
							'completed_at'  => time(),
						]
					);
					$this->result_processor->mark_job_files_as_error( $job_id, 'max_retries_exhausted', 'Scan job failed after maximum retries' );
					$summary['failed_job_ids'][] = $job_id;
				}
				continue;
			}

			$this->record_pending_job( $summary, $job_id, $job, $now );
		}

		return $summary;
	}

	/**
	 * Track pending jobs and the youngest pending age for scan-level poll cadence.
	 *
	 * @param array<string,mixed> $summary
	 * @param string              $job_id
	 * @param array<string,mixed> $job
	 * @param int                 $now
	 * @return void
	 */
	private function record_pending_job( array &$summary, string $job_id, array $job, int $now ): void {
		$summary['pending_job_ids'][] = $job_id;

		$submitted_at = (int) ( $job['submitted_at'] ?? 0 );
		if ( $submitted_at <= 0 ) {
			return;
		}

		$age = max( 0, $now - $submitted_at );
		if ( null === $summary['youngest_pending_age'] || $age < (int) $summary['youngest_pending_age'] ) {
			$summary['youngest_pending_age'] = $age;
		}
	}

	/**
	 * Handle cloud API errors.
	 *
	 * stable client errors (4xx except 408/429) fail the scan immediately
	 * since they will never self-resolve on retry. Transient errors (5xx, network)
	 * accumulate toward MAX_CLOUD_ERRORS before failing.
	 *
	 * @param \WP_Error $error
	 * @param string $context
	 * @return int Counted transient error total; 0 when not counted.
	 */
	private function handle_cloud_error( \WP_Error $error, string $context ): int {
		$error_data = $error->get_error_data( 'api_error' );
		if ( empty( $error_data ) ) {
			$error_data = $error->get_error_data();
		}
		if ( ! is_array( $error_data ) ) {
			$error_data = [];
		}
		$status_code = (int) ( $error_data['status'] ?? 0 );
		$message = $error->get_error_message();
		$error_code = $error->get_error_code();
		$detail = is_array( $error_data['detail'] ?? null ) ? $error_data['detail'] : [];
		$api_error_code = is_string( $detail['error_code'] ?? null ) ? sanitize_key( $detail['error_code'] ) : '';
		$normalized_error_code = $api_error_code !== '' ? $api_error_code : sanitize_key( (string) $error_code );
		$is_rate_limit = ( 429 === $status_code || $error_code === 'rate_limited' || false !== strpos( $message, '429' ) );
		if ( $is_rate_limit ) {
			return 0;
		}

		if ( $normalized_error_code === 'credits_exhausted' ) {
			$this->stop_scan_for_credit_exhaustion( $detail );
			return 0;
		}

		// Map internal context to user-friendly labels for activity log.
		$context_labels = [
			'hash_check'    => 'file analysis',
			'upload'        => 'file upload',
			'poll_results'  => 'results retrieval',
		];
		$label = $context_labels[ $context ] ?? $context;

		// Determine if this is a stable client error that will never resolve.
		// 4xx (except 408 timeout, 425 too early, 429 rate limit) are client-side
		// validation failures. Retrying the same request will produce the same error.
		$is_client_error = ( $status_code >= 400 && $status_code < 500
			&& ! in_array( $status_code, [ 408, 425, 429 ], true ) );

		// Use a professional message for server/network errors; keep API error details as-is.
		if ( in_array( $error_code, [ 'api_server_error', 'api_parse_error', 'http_exception', 'rate_limited' ], true ) ) {
			$summary = sprintf( 'Cloud service unavailable during %s. Will retry automatically.', $label );
		} else {
			$summary = sprintf( 'Cloud %s error: %s%s', $label, $message, $status_code ? " (HTTP $status_code)" : '' );
		}

		$count = \SiteFort\Modules\Scanner\Queue\ScanQueue::record_cloud_error( $summary );

		// Fail immediately on stable client errors, or after threshold on transient errors.
		$should_fail = $is_client_error || ( $count >= self::MAX_CLOUD_ERRORS );

		if ( $should_fail ) {
			$request_id = '';
			if ( isset( $error_data['request_id'] ) ) {
				$request_id = sanitize_text_field( (string) $error_data['request_id'] );
			} elseif ( isset( $detail['request_id'] ) ) {
				$request_id = sanitize_text_field( (string) $detail['request_id'] );
			}
			$host_id = '';
			if ( isset( $error_data['host_id'] ) ) {
				$host_id = sanitize_text_field( (string) $error_data['host_id'] );
			} elseif ( isset( $detail['host_id'] ) ) {
				$host_id = sanitize_text_field( (string) $detail['host_id'] );
			}
			$retryable = ! $is_client_error;

			if ( $is_client_error ) {
				$fail_message = sprintf( 'Cloud %s failed with a non-recoverable error (HTTP %d). Please retry the scan.', $label, $status_code );
			} else {
				$fail_message = 'Cloud service is temporarily unavailable. The scan could not complete after multiple retries. Please try again later.';
			}

			\SiteFort\Modules\Scanner\Queue\ScanQueue::fail(
				$fail_message,
				[
					'error_code'        => $normalized_error_code,
					'stage'             => sanitize_key( $context ),
					'status_code'       => $status_code,
					'request_id'        => $request_id,
					'host_id'           => $host_id,
					'cloud_error_count' => $count,
					'retryable'         => $retryable,
					'technical_message' => sanitize_text_field( $message ),
				]
			);
		}

		return $count;
	}

	/**
	 * Re-queue all files belonging to a cloud job back into pending_upload.
	 *
	 * This is used when the cloud job stays in awaiting_upload/uploading too long, which would otherwise
	 * cause the scan to hang forever.
	 */
	private function requeue_job_files_for_upload( string $job_id ): void {
		global $wpdb;
		$table = $this->table_identifier( $wpdb->prefix . 'sitefort_file_states' );

		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,PluginCheck.Security.DirectDB.UnescapedDBParameter -- Uses a validated plugin-owned table identifier and prepared values.
		$files = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id, state, upload_attempt, file_path FROM {$table}
				WHERE scan_id = %s
				AND cloud_job_id = %s
				AND state IN ('uploaded','scan_requested','uploading')",
				$this->scan_id,
				$job_id
			),
			ARRAY_A
		);
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,PluginCheck.Security.DirectDB.UnescapedDBParameter

		if ( empty( $files ) ) {
			return;
		}

		foreach ( $files as $file ) {
			$attempts = (int) ( $file['upload_attempt'] ?? 0 );

			if ( $attempts >= 3 ) {
				// File has cycled through upload too many times - mark as error to prevent infinite loop.
				$this->fsm->transition(
					(int) $file['id'],
					(string) $file['state'],
					'error',
					[
						'error_code'    => 'upload_cycle_exhausted',
						'error_message' => 'Cloud never acknowledged file receipt after multiple re-uploads',
						'cloud_job_id'  => null,
					]
				);

				CloudResultProcessor::stage_skipped_issue(
					(string) ( $file['file_path'] ?? '' ),
					'Upload failed: cloud never acknowledged file receipt after multiple attempts',
					[ 'error_code' => 'upload_cycle_exhausted', 'upload_attempts' => $attempts ]
				);
				continue;
			}

			$this->fsm->transition(
				(int) $file['id'],
				(string) $file['state'],
				'pending_upload',
				[
					'cloud_job_id' => null,
				]
			);
		}
	}

	/**
	 * File-centric reconciliation: find files stuck in cloud states and resolve them.
	 *
	 * Instead of starting from the cloud_jobs table (which misses orphaned files),
	 * this starts from file_states. Every file in a cloud-related non-terminal state
	 * is checked against its cloud job record and resolved.
	 *
	 * Handles:
	 *  - Files whose cloud job completed/failed but results were not applied
	 *  - Files whose cloud_job_id has no matching job record (orphans)
	 *  - Files with NULL/empty cloud_job_id stuck in uploaded/uploading states
	 *
	 * @return bool True if any remediation was performed.
	 */
	private function reconcile_completed_jobs(): bool {
		$last_reconcile = (int) get_transient( 'sitefort_last_completed_reconcile_' . $this->scan_id );
		if ( ( time() - $last_reconcile ) < 60 ) {
			return false;
		}
		set_transient( 'sitefort_last_completed_reconcile_' . $this->scan_id, time(), 120 );

		global $wpdb;
		$file_table = $this->table_identifier( $wpdb->prefix . 'sitefort_file_states' );
		$jobs_table = $this->table_identifier( $wpdb->prefix . 'sitefort_cloud_jobs' );

		// Step 1: Find ALL files stuck in cloud-related non-terminal states.
		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.PreparedSQL.NotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,PluginCheck.Security.DirectDB.UnescapedDBParameter -- Joined table identifiers are plugin-owned and validated before interpolation; scan_id remains prepared.
		$stuck_files = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT fs.id, fs.file_path, fs.state, fs.cloud_job_id, fs.upload_attempt,
				        cj.status AS job_status, cj.completed_at AS job_completed_at,
				        cj.submitted_at AS job_submitted_at
				FROM {$file_table} fs
				LEFT JOIN {$jobs_table} cj ON fs.cloud_job_id = cj.job_id
				WHERE fs.scan_id = %s
				AND fs.state IN ('uploading','uploaded','scan_requested','scanning')
				LIMIT 500",
				$this->scan_id
			),
			ARRAY_A
		);
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.PreparedSQL.NotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,PluginCheck.Security.DirectDB.UnescapedDBParameter

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

		$did_work = false;

		// Step 2: Group files by their cloud_job_id for batch resolution.
		$by_job = [];
		$orphans = [];
		foreach ( $stuck_files as $file ) {
			$job_id = $file['cloud_job_id'] ?? '';
			if ( empty( $job_id ) || $file['job_status'] === null ) {
				// No job record exists (orphan): job_status is NULL from the LEFT JOIN.
				$orphans[] = $file;
			} else {
				$by_job[ $job_id ][] = $file;
			}
		}

		// Step 3: Handle orphaned files (no matching cloud job record).
		// These files can never be resolved via polling. Re-queue or error immediately.
		foreach ( $orphans as $file ) {
			$attempts = (int) ( $file['upload_attempt'] ?? 0 );
			if ( $attempts < 3 ) {
				$this->fsm->transition(
					(int) $file['id'],
					(string) $file['state'],
					'pending_upload',
					[ 'cloud_job_id' => null ]
				);
			} else {
				$this->fsm->transition(
					(int) $file['id'],
					(string) $file['state'],
					'error',
					[
						'error_code'    => 'orphan_no_job',
						'error_message' => 'File in cloud state but no matching job record exists',
						'cloud_job_id'  => null,
					]
				);
				CloudResultProcessor::stage_skipped_issue(
					$file['file_path'] ?? '(unknown)',
					'Cloud upload failed: no job record found after multiple attempts',
					[ 'error_code' => 'orphan_no_job' ]
				);
			}
			$did_work = true;
		}

		// Step 4: For files with job records, resolve based on job status.
		// Collect job IDs that are terminal locally but still have stuck files.
		$terminal_job_ids = [];
		$active_job_ids = [];
		foreach ( $by_job as $job_id => $files ) {
			$job_status = $files[0]['job_status'] ?? '';
			if ( in_array( $job_status, [ 'completed', 'failed', 'timeout' ], true ) ) {
				$terminal_job_ids[ $job_id ] = $files;
			} else {
				// Job is still active (submitted/processing/queued). Leave for poll_results().
				// But track for timeout detection below.
				$active_job_ids[ $job_id ] = $files;
			}
		}

		// Step 5: Terminal jobs with stuck files. Re-poll cloud for results.
		if ( ! empty( $terminal_job_ids ) ) {
			$job_ids_to_poll = array_keys( $terminal_job_ids );
			$batch_results = $this->cloud_service->poll_batch_results( $job_ids_to_poll );

			foreach ( $terminal_job_ids as $job_id => $files ) {
				$result_data = is_array( $batch_results ) ? ( $batch_results[ $job_id ] ?? null ) : null;
				$cloud_status = is_array( $result_data ) ? ( $result_data['status'] ?? null ) : null;
				$job_status = $files[0]['job_status'];

				if ( 'completed' === $cloud_status ) {
					// Cloud still has results. Apply them.
					$this->result_processor->process_scan_results( $job_id, $result_data['results'] ?? [] );
					$did_work = true;
				} elseif ( in_array( $job_status, [ 'failed', 'timeout' ], true ) ) {
					// Job failed/timed out and cloud has no results. Terminal error.
					$this->result_processor->mark_job_files_as_error( $job_id, 'scan_failed', 'Cloud scan job failed or timed out' );
					$did_work = true;
				} else {
					// Job locally 'completed' but cloud no longer returns results.
					$completed_at = (int) ( $files[0]['job_completed_at'] ?? 0 );
					if ( $completed_at > 0 && ( time() - $completed_at ) > 600 ) {
						$this->result_processor->mark_job_files_as_error( $job_id, 'results_missing', 'Cloud job completed but results could not be retrieved' );
						$did_work = true;
					}
				}
			}
		}

		// Step 6: Active jobs with stuck files. Detect stale active jobs.
		// If a job has been 'submitted'/'processing' for longer than timeout_at allows,
		// poll_results() should handle it. But if poll_results() missed it (no timeout_at
		// set, or status is unexpected like 'queued'), catch it here.
		foreach ( $active_job_ids as $job_id => $files ) {
			$submitted_at = (int) ( $files[0]['job_submitted_at'] ?? 0 );
			$age = $submitted_at > 0 ? ( time() - $submitted_at ) : 0;

			// Jobs older than 15 minutes that are still "active" are stale.
			if ( $submitted_at > 0 && $age > 900 ) {
				$this->result_processor->mark_job_files_as_error( $job_id, 'job_stale', 'Cloud job did not complete within expected timeframe' );
				$this->job_repository->update( $job_id, [
					'status'        => 'failed',
					'error_message' => sprintf( 'Job stale after %ds with no completion', $age ),
					'completed_at'  => time(),
				] );
				$did_work = true;
			}
		}

		return $did_work;
	}
	/**
	 * Check if job has timed out.
	 *
	 * @param array $job Job record.
	 * @return bool True if timed out.
	 */
	private function is_job_timed_out( array $job ): bool {
		$timeout_at = $job['timeout_at'] ?? 0;
		return $timeout_at > 0 && time() > $timeout_at;
	}
	/**
	 * Get active job ID for this scan.
	 *
	 * @return string|null Job ID or null.
	 */
	private function get_active_job_id(): ?string {
		// Check transient first
		$job_id = get_transient( 'sitefort_cloud_job_' . $this->scan_id );
		if ( $job_id ) {
			return $job_id;
		}

		// Check database for active job
		$jobs = $this->job_repository->get_by_scan_id( $this->scan_id );
		foreach ( $jobs as $job ) {
			if ( in_array( $job['status'], [ 'awaiting_upload', 'uploading', 'submitted', 'processing' ], true ) ) {
				return $job['job_id'];
			}
		}

		return null;
	}

	/**
	 * Get malware findings for this scan.
	 *
	 * @return array Malware findings.
	 */
	public function get_findings(): array {
		global $wpdb;

		$table_name = $this->table_identifier( $wpdb->prefix . 'sitefort_file_states' );

		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,PluginCheck.Security.DirectDB.UnescapedDBParameter -- Uses a validated plugin-owned table identifier and a prepared scan_id value.
		$findings = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT * FROM {$table_name}
				WHERE scan_id = %s
				AND state = 'flagged'
				ORDER BY updated_at DESC",
				$this->scan_id
			),
			ARRAY_A
		);
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,PluginCheck.Security.DirectDB.UnescapedDBParameter

		return $findings ?: [];
	}

	/**
	 * Cancel the scan and clean up resources.
	 *
	 * @return void
	 */
	public function cancel(): void {
		// Mark all pending files as error
		global $wpdb;

		$table_name = $this->table_identifier( $wpdb->prefix . 'sitefort_file_states' );

		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,PluginCheck.Security.DirectDB.UnescapedDBParameter -- Uses a validated plugin-owned table identifier and a prepared scan_id value.
		$wpdb->query(
			$wpdb->prepare(
				"UPDATE {$table_name}
				SET state = 'error',
					error_code = 'scan_cancelled',
					error_message = 'Scan was cancelled by user',
					updated_at = %d
				WHERE scan_id = %s
				AND state NOT IN ('completed', 'flagged', 'error', 'verified', 'safe')",
				time(),
				$this->scan_id
			)
		);
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,PluginCheck.Security.DirectDB.UnescapedDBParameter

		// Mark cloud jobs as failed
		$jobs = $this->job_repository->get_by_scan_id( $this->scan_id );

		foreach ( $jobs as $job ) {
			if ( in_array( $job['status'], [ 'submitted', 'processing' ], true ) ) {
				$this->job_repository->update(
					$job['job_id'],
					[
						'status'        => 'failed',
						'error_message' => 'Scan cancelled by user',
						'completed_at'  => time(),
					]
				);
			}
		}

		do_action( 'sitefort_scan_cancelled', $this->scan_id );
	}

	/**
	 * Log scan progress in standard format.
	 *
	 * @param string $phase Phase name for context.
	 */
	/**
	 * Log pressure state transitions to the scan log.
	 * Only logs when the pressure level changes, not on every cycle.
	 *
	 * @param array $pressure Pressure state from AIMD.
	 * @param array $metrics  Raw resource metrics.
	 */
	private function log_pressure_state( array $pressure, array $metrics ): void {
		if ( ! class_exists( '\SiteFort\Modules\Scanner\Service\ScanLogger' ) ) {
			return;
		}

		$level = $pressure['level'] ?? 'normal';

		// Skip if level has not changed since last log.
		if ( $level === self::$last_logged_pressure_level ) {
			return;
		}

		$prev_level = self::$last_logged_pressure_level;
		self::$last_logged_pressure_level = $level;

		$memory = round( (float) ( $metrics['memory'] ?? 0 ), 1 );
		$queue  = (int) ( $metrics['cloud_queue'] ?? 0 );

		if ( $level === 'critical' ) {
			$window = $this->aimd->get_hash_window_for_phase( $pressure, $metrics );
			\SiteFort\Modules\Scanner\Service\ScanLogger::log(
				sprintf(
					'[Resource] CRITICAL pressure - scan throttled to %d files/batch. Memory: %.1f%%, Cloud queue: %d',
					$window, $memory, $queue
				),
				2, 'warning'
			);
		} elseif ( $level === 'moderate' ) {
			$window = $this->aimd->get_hash_window_for_phase( $pressure, $metrics );
			\SiteFort\Modules\Scanner\Service\ScanLogger::log(
				sprintf(
					'[Resource] Moderate pressure - scan throttled to %d files/batch. Memory: %.1f%%, Cloud queue: %d',
					$window, $memory, $queue
				),
				2, 'warning'
			);
		} elseif ( in_array( $prev_level, [ 'critical', 'moderate' ], true ) ) {
			// Only log recovery if we were previously in a constrained state.
			\SiteFort\Modules\Scanner\Service\ScanLogger::log(
				sprintf(
					'[Resource] Pressure cleared - resuming normal scan speed. Memory: %.1f%%',
					$memory
				),
				1, 'info'
			);
		}
	}

	private function log_scan_progress( string $phase ): void {
		if ( ! class_exists( '\SiteFort\Modules\Scanner\Service\ScanLogger' ) ) {
			return;
		}

		$progress = $this->fsm->get_scan_progress( $this->scan_id );
		$by_state = $progress['by_state'] ?? [];

		// Terminal states = processed
		$processed = (
			( $by_state['completed'] ?? 0 ) +
			( $by_state['flagged'] ?? 0 ) +
			( $by_state['error'] ?? 0 ) +
			( $by_state['verified'] ?? 0 ) +
			( $by_state['safe'] ?? 0 ) +
			( $by_state['scan_skipped'] ?? 0 )
		);

		$total = $progress['total'] ?? 0;

		$cloud_pending = (
			( $by_state['pending_upload'] ?? 0 ) +
			( $by_state['uploading'] ?? 0 ) +
			( $by_state['uploaded'] ?? 0 ) +
			( $by_state['scan_requested'] ?? 0 ) +
			( $by_state['scanning'] ?? 0 )
		);
		$processed_label = function_exists( 'number_format_i18n' ) ? number_format_i18n( $processed ) : (string) $processed;
		$total_label     = function_exists( 'number_format_i18n' ) ? number_format_i18n( $total ) : (string) $total;
		$remaining       = max( 0, $total - $processed );
		$remaining_label = function_exists( 'number_format_i18n' ) ? number_format_i18n( $remaining ) : (string) $remaining;
		$cloud_label     = function_exists( 'number_format_i18n' ) ? number_format_i18n( $cloud_pending ) : (string) $cloud_pending;

		$msg = sprintf(
			'Scan Progress: %s/%s files checked.',
			$processed_label,
			$total_label
		);

		if ( $remaining > 0 ) {
			if ( $cloud_pending > 0 ) {
				$msg .= sprintf(
					' %s file(s) remaining, including %s awaiting cloud processing.',
					$remaining_label,
					$cloud_label
				);
			} else {
				$msg .= sprintf(
					' %s file(s) remaining.',
					$remaining_label
				);
			}
		} else {
			$msg .= ' Finalizing scan results.';
		}

		\SiteFort\Modules\Scanner\Service\ScanLogger::log( $msg, 1, 'info' );
	}

	/**
	 * Quote a table identifier after validating its shape.
	 */
	private function table_identifier( string $table ): string {
		if ( $table === '' || strlen( $table ) > 64 || ! preg_match( '/^[A-Za-z0-9_]+$/', $table ) ) {
			throw new \InvalidArgumentException( 'Invalid table identifier.' );
		}

		return '`' . $table . '`';
	}

}
