<?php
/**
 * Costudio_Runner — Async state-machine runner.
 *
 * Processes productions one agent at a time in the background.
 * Dispatch uses a hybrid approach:
 *  1. Non-blocking Loopback HTTP (REST) for immediate execution.
 *  2. WP-Cron as a single-shot fallback if the loopback never arrives.
 *
 * On provider/pipeline exceptions, the production is marked as error and the
 * user can retry via the UI. Silent PHP deaths are detected by
 * Costudio_Housekeeping's watchdog, which marks the production as a retryable
 * error instead of auto-running the same agent twice.
 *
 * @package Contentosapp_Studio
 * @since   3.2.0
 */

defined( 'ABSPATH' ) || exit;
// phpcs:disable Squiz.PHP.DiscouragedFunctions.Discouraged,WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.PreparedSQL.NotPrepared,WordPress.DB.DirectDatabaseQuery,PluginCheck.Security.DirectDB.UnescapedDBParameter -- Runner: set_time_limit() required for long AI runs; DB queries on plugin-owned tables only.

class Costudio_Runner {

	const CRON_HOOK = 'costudio_pipeline_cron_run';

	private const PIPELINE_LOG_MAX = 40;
	private const HANDOFF_RECOVERY_AFTER_SECONDS = 30;
	private const HANDOFF_RECOVERY_DEBOUNCE_SECONDS = 30;

	// ── Logging & Diag ──────────────────────────────────────────────────────

	/**
	 * Append an entry to the pipeline activity log (transient; shown in the production UI).
	 * Supports structured data for color-coded levels and agent context.
	 *
	 * @param int         $production_id
	 * @param string      $message
	 * @param string      $level         'info', 'ok', 'warn', 'err'
	 * @param int|null    $agent         Agent number (1-7)
	 */
	public static function pipeline_log_append( int $production_id, string $message, string $level = 'info', ?int $agent = null ): void {
		$key   = 'costudio_plog_' . $production_id;
		$lines = get_transient( $key );
		if ( ! is_array( $lines ) ) {
			$lines = [];
		}

		$entry = [
			'ts'    => time(),
			'level' => $level,
			'agent' => $agent,
			'msg'   => $message,
		];
		$lines[] = $entry;

		if ( count( $lines ) > self::PIPELINE_LOG_MAX ) {
			$lines = array_slice( $lines, -self::PIPELINE_LOG_MAX );
		}

		set_transient( $key, $lines, 2 * HOUR_IN_SECONDS );
		set_transient( 'costudio_plog_last_' . $production_id, time(), 2 * HOUR_IN_SECONDS );

		if ( class_exists( 'Costudio_Logger' ) && method_exists( 'Costudio_Logger', 'pipeline_event' ) ) {
			Costudio_Logger::pipeline_event( $production_id, $message, $level, $agent, (int) $entry['ts'] );
		}
	}

	/**
	 * @return string[]
	 */
	public static function get_pipeline_log( int $production_id ): array {
		if ( class_exists( 'Costudio_Logger' ) && method_exists( 'Costudio_Logger', 'query_pipeline_events' ) ) {
			$persisted = Costudio_Logger::query_pipeline_events( $production_id );
			if ( ! empty( $persisted ) ) {
				return $persisted;
			}
		}

		$lines = get_transient( 'costudio_plog_' . $production_id );
		return is_array( $lines ) ? $lines : [];
	}

	/**
	 * Get diagnostic data for the pipeline.
	 */
	public static function get_pipeline_diag( int $production_id ): array {
		$production = Costudio_Production::get_by_id( $production_id );
		if ( ! $production ) {
			return [];
		}

		$cron_key = self::CRON_HOOK;
		$crons    = _get_cron_array();
		$pending_token_age = 0;
		$has_pending_token = false;
		$primed_at         = (int) get_transient( "costudio_bg_at_{$production_id}" );
		$stored_token      = get_transient( "costudio_bg_{$production_id}" );
		if ( is_string( $stored_token ) && '' !== $stored_token ) {
			$has_pending_token = true;
			$pending_token_age = $primed_at > 0 ? max( 0, time() - $primed_at ) : 0;
		}

		if ( $crons ) {
			foreach ( $crons as $timestamp => $cronhooks ) {
				if ( isset( $cronhooks[ $cron_key ] ) ) {
					foreach ( $cronhooks[ $cron_key ] as $sig => $data ) {
						if ( isset( $data['args'][0] ) && (int) $data['args'][0] === $production_id ) {
							$has_pending_token = true;
							if ( $primed_at <= 0 ) {
								$pending_token_age = max( 0, time() - $timestamp );
							}
							break 2;
						}
					}
				}
			}
		}

		$version       = $production->get_pipeline_version();
		$agent_qty     = ( 7 === $version ) ? 7 : 6;
		$current_agent = (int) ( $production->current_agent ?? 0 );
		$all_agents_done = true;
		for ( $i = 1; $i <= $agent_qty; $i++ ) {
			$col = method_exists( $production, 'map_step_to_column' )
				? $production->map_step_to_column( $i )
				: sprintf( 'output_%02d', $i );
			if ( empty( $production->$col ) ) {
				$all_agents_done = false;
				break;
			}
		}
		$current_stage = $current_agent ? 'Agent ' . $current_agent : 'boot';
		$phase         = 'agent';

		if ( 'running' === $production->status && ( $current_agent > $agent_qty || $all_agents_done ) ) {
			$current_stage = __( 'Publishing to WordPress', 'contentosapp-studio' );
			$phase         = 'publishing';
		} elseif ( 'completed' === $production->status ) {
			$current_stage = __( 'Completed', 'contentosapp-studio' );
			$phase         = 'completed';
		}

		$publication_diag = class_exists( 'Costudio_Publication_Jobs' )
			? Costudio_Publication_Jobs::diagnostics( $production_id )
			: array();
		$handoff_age = empty( $production->current_agent_started_at )
			? self::handoff_age_seconds( $production )
			: 0;

		return [
			'runner_state'          => $production->status,
			'current_stage'         => $current_stage,
			'phase'                 => $phase,
			'has_pending_token'     => $has_pending_token,
			'pending_token_source'  => get_transient( "costudio_bg_src_{$production_id}" ) ?: 'cron',
			'pending_token_age'     => $pending_token_age,
			'last_heartbeat_at'     => $production->heartbeat ? strtotime( $production->heartbeat . ' UTC' ) : 0,
			'run_lock_active'       => self::get_lock_age( $production_id ) > 0,
			'run_lock_age'          => self::get_lock_age( $production_id ),
			'last_log_at'           => get_transient( 'costudio_plog_last_' . $production_id ) ?: 0,
			'last_dispatch_method'  => 'WP-Cron',
			'last_dispatch_result'  => 'scheduled',
			'handoff_waiting'       => 'running' === $production->status && $current_agent >= 1 && $current_agent <= $agent_qty && empty( $production->current_agent_started_at ),
			'handoff_age'           => $handoff_age,
			'publication_job'       => $publication_diag,
		];
	}

	private static function get_lock_age( int $production_id ): int {
		$value = get_option( self::lock_key_for( $production_id ), '' );
		if ( ! is_string( $value ) || strpos( $value, '|' ) === false ) {
			return 0;
		}

		$parts = explode( '|', $value, 2 );
		$ts    = (int) ( $parts[1] ?? 0 );
		return $ts > 0 ? max( 0, time() - $ts ) : 0;
	}

	// ── Background Dispatch ──────────────────────────────────────────────────

	/**
	 * Save a background token for a production.
	 */
	public static function prime_background_token( int $production_id, string $token, string $source = '' ): void {
		set_transient( "costudio_bg_{$production_id}", $token, 2 * HOUR_IN_SECONDS );
		set_transient( "costudio_bg_at_{$production_id}", time(), 2 * HOUR_IN_SECONDS );
		if ( $source ) {
			set_transient( "costudio_bg_src_{$production_id}", $source, 2 * HOUR_IN_SECONDS );
		}
	}

	/**
	 * Recover the safe boundary between two agents when the next runner never starts.
	 *
	 * At this boundary current_agent has already advanced, but
	 * current_agent_started_at is still empty. Therefore no provider request for
	 * the selected agent can have been dispatched yet. Rotating the one-time
	 * background token is safe; the runner lock remains the final single-flight
	 * guard if the original loopback arrives concurrently.
	 */
	public static function ensure_handoff_progress( int $production_id, string $source = 'poll' ): bool {
		$production = Costudio_Production::get_by_id( $production_id );
		if ( ! self::is_recoverable_handoff( $production ) ) {
			return false;
		}

		$threshold = max(
			10,
			(int) apply_filters( 'costudio_handoff_recovery_after_seconds', self::HANDOFF_RECOVERY_AFTER_SECONDS, $production_id )
		);
		$handoff_age = self::handoff_age_seconds( $production );
		if ( $handoff_age < $threshold || self::get_lock_age( $production_id ) > 0 ) {
			return false;
		}

		$agent = (int) $production->current_agent;
		if ( ! self::acquire_handoff_recovery_debounce( $production_id, $agent ) ) {
			return false;
		}

		// Re-read after winning the debounce. The original runner may have started
		// between the first read and this lock acquisition.
		$production = Costudio_Production::get_by_id( $production_id );
		if ( ! self::is_recoverable_handoff( $production ) || self::get_lock_age( $production_id ) > 0 ) {
			delete_option( "costudio_handoff_recovery_{$production_id}" );
			return false;
		}

		$agent       = (int) $production->current_agent;
		$handoff_age = self::handoff_age_seconds( $production );
		$primed_at   = (int) get_transient( "costudio_bg_at_{$production_id}" );
		$context     = array(
			'production_id'    => $production_id,
			'agent_id'         => $agent,
			'handoff_age'      => $handoff_age,
			'pending_token_age' => $primed_at > 0
				? max( 0, time() - $primed_at )
				: 0,
			'source'           => sanitize_key( $source ),
			'php_sapi'         => PHP_SAPI,
		);

		if ( class_exists( 'Costudio_Logger' ) ) {
			Costudio_Logger::warn( 'runner.handoff_stalled', $context, $production_id, $agent );
		}
		if ( function_exists( 'costudio_log' ) ) {
			costudio_log( 'runner.handoff_stalled', $context, 'warning' );
		}

		$token = wp_generate_password( 32, false );
		self::prime_background_token( $production_id, $token, 'handoff_recovery_' . sanitize_key( $source ) );
		self::kick_off_background( $production_id, $token );
		self::pipeline_log_append(
			$production_id,
			__( 'The next agent did not start on the first dispatch. Contentosapp Studio recovered the handoff automatically.', 'contentosapp-studio' ),
			'info',
			$agent
		);

		if ( class_exists( 'Costudio_Logger' ) ) {
			Costudio_Logger::info( 'runner.handoff_recovery_kicked', $context, $production_id, $agent );
		}
		if ( function_exists( 'costudio_log_debug' ) ) {
			costudio_log_debug( 'runner.handoff_recovery_kicked', $context );
		}

		return true;
	}

	/** Recover handoff gaps even when no production detail page is open. */
	public static function recover_stalled_handoffs( int $limit = 20, string $source = 'watchdog' ): int {
		global $wpdb;

		$table = $wpdb->prefix . Costudio_Production::TABLE;
		$rows  = $wpdb->get_col(
			$wpdb->prepare(
				"SELECT id FROM {$table}
				 WHERE status = 'running'
				   AND current_agent BETWEEN 1 AND 7
				   AND (current_agent_started_at IS NULL OR current_agent_started_at = '')
				 ORDER BY updated_at ASC
				 LIMIT %d",
				max( 1, $limit )
			)
		);

		$recovered = 0;
		foreach ( $rows ?: array() as $production_id ) {
			if ( self::ensure_handoff_progress( (int) $production_id, $source ) ) {
				$recovered++;
			}
		}

		return $recovered;
	}

	private static function is_recoverable_handoff( $production ): bool {
		if ( ! $production || 'running' !== $production->status || ! empty( $production->current_agent_started_at ) ) {
			return false;
		}

		$agent     = (int) ( $production->current_agent ?? 0 );
		$agent_qty = 7 === $production->get_pipeline_version() ? 7 : 6;
		if ( $agent < 1 || $agent > $agent_qty ) {
			return false;
		}
		$output_column = $production->map_step_to_column( $agent );
		if ( ! $output_column || ! empty( $production->$output_column ) ) {
			// A selected agent with an existing output is not a clean pre-dispatch
			// boundary. Leave that inconsistent state to explicit reconciliation.
			return false;
		}

		$id = (int) $production->id;
		if ( class_exists( 'Costudio_Publication_Jobs' ) && Costudio_Publication_Jobs::has_live_job( $id ) ) {
			return false;
		}
		if ( class_exists( 'Costudio_Remote_Jobs' ) && Costudio_Remote_Jobs::has_live_job( $id ) ) {
			return false;
		}
		if ( class_exists( 'Costudio_BYOK_Jobs' ) && ( Costudio_BYOK_Jobs::has_live_job( $id ) || Costudio_BYOK_Jobs::has_pending_application( $id ) ) ) {
			return false;
		}

		return true;
	}

	private static function handoff_age_seconds( $production ): int {
		if ( ! empty( $production->heartbeat ) ) {
			$beat = strtotime( (string) $production->heartbeat . ' UTC' );
			if ( $beat ) {
				return max( 0, time() - $beat );
			}
		}

		if ( ! empty( $production->updated_at ) ) {
			$updated_gmt = get_gmt_from_date( (string) $production->updated_at );
			$updated     = strtotime( $updated_gmt . ' UTC' );
			if ( $updated ) {
				return max( 0, time() - $updated );
			}
		}

		return 0;
	}

	private static function acquire_handoff_recovery_debounce( int $production_id, int $agent ): bool {
		global $wpdb;

		$key   = "costudio_handoff_recovery_{$production_id}";
		$value = $agent . '|' . time();
		$added = $wpdb->query(
			$wpdb->prepare(
				"INSERT IGNORE INTO {$wpdb->options} (option_name, option_value, autoload) VALUES (%s, %s, 'no')",
				$key,
				$value
			)
		);
		if ( 1 === (int) $added ) {
			wp_cache_delete( $key, 'options' );
			return true;
		}

		$existing = (string) $wpdb->get_var(
			$wpdb->prepare( "SELECT option_value FROM {$wpdb->options} WHERE option_name = %s", $key )
		);
		$parts       = explode( '|', $existing, 2 );
		$old_agent   = (int) ( $parts[0] ?? 0 );
		$existing_ts = (int) ( $parts[1] ?? 0 );
		$debounce    = max(
			10,
			(int) apply_filters( 'costudio_handoff_recovery_debounce_seconds', self::HANDOFF_RECOVERY_DEBOUNCE_SECONDS, $production_id )
		);
		if ( $old_agent === $agent && $existing_ts > 0 && ( time() - $existing_ts ) < $debounce ) {
			return false;
		}

		$updated = $wpdb->query(
			$wpdb->prepare(
				"UPDATE {$wpdb->options} SET option_value = %s WHERE option_name = %s AND option_value = %s",
				$value,
				$key,
				$existing
			)
		);
		wp_cache_delete( $key, 'options' );

		return 1 === (int) $updated;
	}

	/**
	 * Trigger the background process.
	 */
	public static function kick_off_background( int $production_id, string $token ): void {
		self::pipeline_log_append( $production_id, __('A despachar runner em segundo plano…', 'contentosapp-studio'), 'info' );

		// Primary path: non-blocking loopback HTTP to the REST endpoint.
		self::fire_background_job( $production_id, $token );

		// Fallback: schedule a single WP-Cron event with the same token. The
		// token lives as a transient and is consumed by whichever path fires
		// first; the second finds it gone and returns.
		wp_schedule_single_event( time(), self::CRON_HOOK, [ $production_id, $token ] );
		if ( ! ( defined( 'DISABLE_WP_CRON' ) && DISABLE_WP_CRON ) && function_exists( 'spawn_cron' ) ) {
			spawn_cron();
		}
	}

	/** Schedule a read-only status pass without opening a tight loopback chain. */
	public static function schedule_background_check( int $production_id, string $token, int $delay = 10 ): void {
		wp_schedule_single_event( time() + max( 5, $delay ), self::CRON_HOOK, [ $production_id, $token ] );
	}

	/**
	 * Consume a terminal durable BYOK result and hand the production to the
	 * next agent. When called by the CLI worker, run the short local handoff in
	 * the same process so it does not depend on a later WP-Cron beat.
	 */
	public static function resume_durable_result( int $production_id, string $source = 'worker_terminal' ): void {
		$production = Costudio_Production::get_by_id( $production_id );
		if ( ! $production || in_array( $production->status, [ 'completed', 'error', 'paused', 'cancelled' ], true ) ) {
			return;
		}

		$token = wp_generate_password( 32, false );
		self::prime_background_token( $production_id, $token, 'durable_byok_' . sanitize_key( $source ) );
		if ( class_exists( 'Costudio_Logger' ) ) {
			Costudio_Logger::info(
				'byok_jobs.local_handoff_started',
				array(
					'production_id' => $production_id,
					'agent_id'      => max( 1, (int) ( $production->current_agent ?? 1 ) ),
					'source'        => sanitize_key( $source ),
					'php_sapi'      => PHP_SAPI,
				),
				$production_id,
				max( 1, (int) ( $production->current_agent ?? 1 ) )
			);
		}

		if ( 'cli' === PHP_SAPI ) {
			self::run( $production_id, $token );
			return;
		}

		self::kick_off_background( $production_id, $token );
	}

	/**
	 * Dispatch a non-blocking HTTP POST to the background REST endpoint.
	 */
	public static function fire_background_job( int $production_id, string $secret_token ): void {
		$url = apply_filters( 'costudio_background_job_url', rest_url( 'costudio/v1/run-background' ), $production_id );

		wp_remote_post(
			$url,
			[
				'blocking'  => false,
				'timeout'   => 0.01,
				// phpcs:ignore WordPress.NamingConventions.PrefixAllGlobals.NonPrefixedHooknameFound -- WordPress core loopback SSL verification filter.
				'sslverify' => apply_filters( 'https_local_ssl_verify', false ),
				'headers'   => [ 'Content-Type' => 'application/x-www-form-urlencoded' ],
				'body'      => [
					'production_id' => $production_id,
					'secret_token'  => $secret_token,
				],
			]
		);
	}

	// ── Runner Engine ────────────────────────────────────────────────────────

	/** Stale-lock cutoff: a lock older than this is assumed orphaned. */
	private const LOCK_STALE_AFTER_SECONDS = 900;

	/**
	 * Try to acquire a distributed lock for this run via MySQL INSERT IGNORE,
	 * which is atomic at the DB level (relies on the unique key on
	 * wp_options.option_name).
	 *
	 * `add_option()` cannot be used here: WP core uses INSERT ... ON DUPLICATE
	 * KEY UPDATE, so it never fails when the option already exists — both
	 * racing workers would see "success" and both would proceed.
	 *
	 * Stale locks (older than LOCK_STALE_AFTER_SECONDS without a release) are
	 * assumed to belong to a crashed worker and are taken over. The lock value
	 * encodes the token + acquisition timestamp as `"<token>|<unix_ts>"`.
	 *
	 * @return bool True if the lock was acquired, false if another worker holds it.
	 */
	private static function try_acquire_lock( int $production_id, string $secret_token ): bool {
		global $wpdb;

		$lock_key   = self::lock_key_for( $production_id );
		$lock_value = $secret_token . '|' . time();

		$inserted = $wpdb->query( $wpdb->prepare(
			"INSERT IGNORE INTO {$wpdb->options} (option_name, option_value, autoload) VALUES (%s, %s, 'no')",
			$lock_key,
			$lock_value
		) );

		if ( $inserted === 1 ) {
			wp_cache_delete( $lock_key, 'options' );
			return true;
		}

		// Lock already exists — check if it's stale (crashed worker).
		$existing = $wpdb->get_var( $wpdb->prepare(
			"SELECT option_value FROM {$wpdb->options} WHERE option_name = %s",
			$lock_key
		) );

		if ( ! $existing || strpos( $existing, '|' ) === false ) {
			return false;
		}

		$parts       = explode( '|', $existing, 2 );
		$existing_ts = (int) ( $parts[1] ?? 0 );

		if ( $existing_ts <= 0 || ( time() - $existing_ts ) <= self::LOCK_STALE_AFTER_SECONDS ) {
			// Active lock held by another worker — back off.
			return false;
		}

		// Stale lock — take it over only if nobody changed it meanwhile.
		$updated = $wpdb->query( $wpdb->prepare(
			"UPDATE {$wpdb->options} SET option_value = %s WHERE option_name = %s AND option_value = %s",
			$lock_value,
			$lock_key,
			$existing
		) );
		wp_cache_delete( $lock_key, 'options' );

		return 1 === (int) $updated;
	}

	/**
	 * Release the distributed lock. Safe to call multiple times.
	 */
	private static function release_lock( int $production_id ): void {
		delete_option( self::lock_key_for( $production_id ) );
	}

	private static function lock_key_for( int $production_id ): string {
		return "costudio_runner_lock_{$production_id}";
	}

	/**
	 * Main execution entry point (decoupled from HTTP request).
	 */
	public static function run( int $production_id, string $secret_token ): void {
		// Cheap pre-check: bail early on obviously invalid tokens before touching the DB lock.
		$stored = get_transient( "costudio_bg_{$production_id}" );
		if ( $stored !== $secret_token ) {
			return;
		}

		// Atomic distributed lock — prevents loopback HTTP and WP-Cron from
		// running the same agent in parallel under distributed object cache
		// (Redis/Memcached) where transient consumption can race.
		if ( ! self::try_acquire_lock( $production_id, $secret_token ) ) {
			costudio_log_debug( 'runner.lock_contention', [ 'production_id' => $production_id ] );
			return;
		}

		try {
			// Consume the transient (cosmetic now — the real guard is the lock above).
			$token_source = (string) get_transient( "costudio_bg_src_{$production_id}" );
			delete_transient( "costudio_bg_{$production_id}" );
			delete_transient( "costudio_bg_at_{$production_id}" );
			delete_transient( "costudio_bg_src_{$production_id}" );

			// phpcs:ignore WordPress.PHP.NoSilencedErrors
			@set_time_limit( 0 );
			if ( function_exists( 'ignore_user_abort' ) ) {
				ignore_user_abort( true );
			}

			$fresh = Costudio_Production::get_by_id( $production_id );
			if ( ! $fresh || in_array( $fresh->status, [ 'completed', 'error', 'paused', 'cancelled' ], true ) ) {
				self::release_lock( $production_id );
				return;
			}

			$fresh->status    = 'running';
			$fresh->heartbeat = current_time( 'mysql', true );
			$fresh->save();

			$agent_num = (int) ( $fresh->current_agent ?? 1 );
			if ( $agent_num < 1 ) {
				$agent_num = 1;
			}

			$version   = $fresh->get_pipeline_version();
			$agent_qty = ( 7 === $version ) ? 7 : 6;

			if ( $agent_num > $agent_qty ) {
				self::pipeline_log_append( $production_id, __('A finalizar e publicar no WordPress…', 'contentosapp-studio'), 'info' );
				if ( class_exists( 'Costudio_Publication_Jobs' ) && Costudio_Publication_Jobs::enabled() ) {
					$row = Costudio_Publication_Jobs::ensure_queued( $production_id, 'runner_boundary' );
					if ( $row ) {
						self::release_lock( $production_id );
						Costudio_Publication_Jobs::ensure_progress( $production_id, 'runner_boundary' );
						self::pipeline_log_append( $production_id, __( 'Publicação adicionada à fila durável. O processo continuará automaticamente.', 'contentosapp-studio' ), 'info' );
						return;
					}

					// A migration/engine problem must not strand a production. Continue
					// through the pre-existing bounded publication path and leave a clear
					// operational diagnostic for support.
					self::pipeline_log_append( $production_id, __( 'A fila durável de publicação não estava disponível; usando a publicação compatível desta instalação.', 'contentosapp-studio' ), 'warn' );
					if ( function_exists( 'costudio_log' ) ) {
						costudio_log( 'publication_jobs.queue_unavailable_fallback', array( 'production_id' => $production_id ), 'warn' );
					}
				}
			} else {
				// Claim once per attempt. ContentOS Auto may re-enter this runner
				// every ~45 seconds while polling the same remote job; the
				// conditional claim preserves the original boundary.
				$claimed_started_at = Costudio_Production::claim_current_agent_started_at( $production_id, $agent_num );
				delete_option( "costudio_handoff_recovery_{$production_id}" );
				if ( 0 === strpos( $token_source, 'handoff_recovery_' ) && class_exists( 'Costudio_Logger' ) ) {
					$recovery_context = array(
							'production_id' => $production_id,
							'agent_id'      => $agent_num,
							'started_at'    => (string) $claimed_started_at,
							'source'        => sanitize_key( substr( $token_source, strlen( 'handoff_recovery_' ) ) ),
							'php_sapi'      => PHP_SAPI,
						);
					Costudio_Logger::info(
						'runner.handoff_recovered',
						$recovery_context,
						$production_id,
						$agent_num
					);
					if ( function_exists( 'costudio_log_debug' ) ) {
						costudio_log_debug( 'runner.handoff_recovered', $recovery_context );
					}
				}

				self::pipeline_log_append(
					$production_id,
					sprintf(
						/* translators: %02d: two-digit agent number. */
						__( 'A processar o Agente %02d…', 'contentosapp-studio' ),
						$agent_num
					),
					'info',
					$agent_num
				);
			}

			try {
				$pipeline   = new Costudio_Pipeline();
				if ( $agent_num <= $agent_qty ) {
					$pipeline->defer_to_durable_worker_if_needed( $production_id, $agent_num );
				}
				$next_agent = $pipeline->run_step( $agent_num, $production_id );

				// Check again if user paused or cancelled while the agent was running.
				// This prevents scheduling the next step if the current one finished
				// just as the user clicked "Pause" or "Cancel".
				$fresh = Costudio_Production::get_by_id( $production_id );
				if ( $fresh && in_array( $fresh->status, [ 'error', 'paused', 'cancelled', 'pending_upgrade' ], true ) ) {
					self::release_lock( $production_id );
					return;
				}

				// Step finished — the per-token lock has done its job. Release
				// it BEFORE dispatching the next agent's loopback HTTP request,
				// otherwise that request races back into run() and finds the
				// lock still held, bailing out and stalling the pipeline.
				self::release_lock( $production_id );

				if ( $next_agent ) {
					// Check if user paused or cancelled the pipeline while the agent was running
					$fresh = Costudio_Production::get_by_id( $production_id );
					if ( $fresh && in_array( $fresh->status, [ 'error', 'paused', 'cancelled', 'pending_upgrade' ], true ) ) {
						return;
					}

					// ── Manual Review Guard ──────────────────────────────────
					// If mode is 'manual', we halt after each step and wait for
					// user approval. We switch status to 'awaiting_approval'.
					$inputs = json_decode( $fresh->inputs ?? '{}', true );
					if ( 'manual' === ( $inputs['execution_mode'] ?? 'auto' ) ) {
						$transitioned = Costudio_Production::transition_to_agent(
							$production_id,
							$agent_num,
							(int) $next_agent,
							'awaiting_approval'
						);
						if ( $transitioned && class_exists( 'Costudio_BYOK_Gateway' ) ) {
							Costudio_BYOK_Gateway::mark_applied( $production_id, $agent_num );
						}

						self::pipeline_log_append( 
							$production_id, 
							sprintf(
								/* translators: %1$02d: completed two-digit agent number, %2$02d: next two-digit agent number. */
								__( 'Agente %1$02d concluído. Aguardando aprovação para despachar Agente %2$02d.', 'contentosapp-studio' ),
								$agent_num,
								$next_agent
							), 
							'info', 
							$next_agent 
						);
						return;
					}

					$next_is_publish_step = (int) $next_agent > $agent_qty;

					// Log the "agents done → publishing" banner ONLY on the first
					// transition (agent → publish), not on each chunked publish
					// re-dispatch (publish step returning itself, one per image).
					if ( $next_is_publish_step && (int) $agent_num <= $agent_qty ) {
						self::pipeline_log_append(
							$production_id,
							__('Agentes concluídos. A preparar referências, imagens, metadados e publicação no WordPress…', 'contentosapp-studio'),
							'info'
						);
					}

					$transitioned = Costudio_Production::transition_to_agent(
						$production_id,
						$agent_num,
						(int) $next_agent
					);
					if ( $transitioned && class_exists( 'Costudio_BYOK_Gateway' ) ) {
						Costudio_BYOK_Gateway::mark_applied( $production_id, $agent_num );
					}

					if ( $transitioned && $next_is_publish_step && class_exists( 'Costudio_Publication_Jobs' ) && Costudio_Publication_Jobs::enabled() ) {
						$row = Costudio_Publication_Jobs::ensure_queued( $production_id, 'agent_handoff' );
						if ( $row ) {
							Costudio_Publication_Jobs::ensure_progress( $production_id, 'agent_handoff' );
							self::pipeline_log_append( $production_id, __( 'Publicação adicionada à fila durável. O processo continuará automaticamente.', 'contentosapp-studio' ), 'info' );
						}
						return;
					}

					$new_token = wp_generate_password( 32, false );
					self::prime_background_token( $production_id, $new_token, 'chain' );

					// A terminal durable BYOK result is already running inside the
					// dedicated CLI worker. Start the short next-agent boundary in
					// this same process so creating the next durable job cannot be
					// lost to a loopback HTTP gap. Publishing remains on the normal
					// background path because it is not part of the BYOK queue.
					$continue_byok_in_cli = 'cli' === PHP_SAPI
						&& (int) $next_agent <= $agent_qty
						&& 'byok' === (string) ( $inputs['execution_ai_mode'] ?? $inputs['ai_mode'] ?? '' )
						&& class_exists( 'Costudio_BYOK_Jobs' )
						&& Costudio_BYOK_Jobs::routes_new_jobs();

					if ( $continue_byok_in_cli ) {
						self::run( $production_id, $new_token );
					} else {
						self::kick_off_background( $production_id, $new_token );
					}
				} else {
					// CRITICAL: run_step() can return null in two distinct cases:
					//   (a) pipeline finished normally (last agent done) → mark completed.
					//   (b) the agent called set_error() and aborted → DB.status is 'error'
					//       and we must NOT overwrite that with 'completed'.
					// The in-memory production was loaded with status='running' before
					// run_step() executed; saving it blindly would clobber the error
					// status that set_error() persisted directly to the DB.
					$fresh = Costudio_Production::get_by_id( $production_id );
					if ( ! $fresh ) {
						return;
					}
					Costudio_Production::clear_current_agent_timing( $production_id, $agent_num );

					if ( $fresh->status === 'error' ) {
						self::pipeline_log_append(
							$production_id,
							sprintf(
								/* translators: %s: production error message. */
								__( 'Produção interrompida com erro: %s', 'contentosapp-studio' ),
								(string) ( $fresh->error_message ?? 'unknown' )
							),
							'err',
							$agent_num
						);
						return;
					}

					if ( in_array( $fresh->status, [ 'paused', 'cancelled' ], true ) ) {
						return;
					}

					if ( $fresh->status !== 'completed' ) {
						Costudio_Production::mark_pipeline_completed( $production_id );
						$fresh = Costudio_Production::get_by_id( $production_id );
					}

					if ( ! $fresh ) {
						return;
					}

					self::pipeline_log_append( $production_id, __( 'Produção concluída.', 'contentosapp-studio' ), 'ok' );
					$inputs = json_decode( $fresh->inputs ?? '{}', true );
					if ( ! is_array( $inputs ) ) {
						$inputs = [];
					}

					if (
						'contentos' === (string) ( $inputs['execution_ai_mode'] ?? $inputs['ai_mode'] ?? '' )
						&& class_exists( 'Costudio_Standalone_License' )
						&& method_exists( 'Costudio_Standalone_License', 'sync_remote_usage' )
					) {
						Costudio_Standalone_License::sync_remote_usage( true );
					}

					// ── Telemetry: Production Completed ──────────────────────────
					if ( class_exists( 'Costudio_Telemetry' ) ) {
						$totals = Costudio_Usage::get_production_totals( $production_id );

						$cts = function_exists( 'costudio_settings' ) ? costudio_settings() : null;
						$img_qty = $cts ? (int) $cts->get( 'costudio_image_quantity', 2 ) : 2;
						$img_ai_mode = (string) ( $inputs['execution_ai_mode'] ?? $inputs['ai_mode'] ?? ( $cts ? $cts->get( 'ai_mode', '' ) : '' ) );
						$img_qty = class_exists( 'Costudio_Standalone_Settings' )
							? Costudio_Standalone_Settings::clamp_image_quantity( $img_qty, $img_ai_mode )
							: max( 1, min( 3, $img_qty ) );
						if ( ( $cts ? (string) $cts->get( 'costudio_image_mode', 'generate' ) : 'generate' ) === 'none' ) {
							$img_qty = 0;
						}

						$production_uid = Costudio_Production::tracking_uid_for( $production_id );
						$destination_context = [];
						$destination_site_key = sanitize_key( (string) ( $fresh->site_key ?? '' ) );
						$destination_name = $destination_site_key;
						$destination_url  = '';
						if ( '__local__' === $destination_site_key ) {
							$destination_url  = home_url();
							$destination_name = get_bloginfo( 'name' ) ?: $destination_site_key;
						} elseif ( '' !== $destination_site_key && function_exists( 'costudio_sites' ) ) {
							$sites_repo = costudio_sites();
							$site_cfg   = $sites_repo ? $sites_repo->get( $destination_site_key ) : null;
							if ( is_array( $site_cfg ) ) {
								$destination_url  = (string) ( $site_cfg['url'] ?? '' );
								$destination_name = function_exists( 'costudio_site_display_name' )
									? costudio_site_display_name( $site_cfg, $destination_site_key )
									: (string) ( $site_cfg['name'] ?? $site_cfg['label'] ?? $destination_site_key );
							}
						}
						$destination_url = costudio_safe_http_url( rtrim( $destination_url, '/' ) );
						$wp_post_url     = esc_url_raw( (string) ( $fresh->wp_post_url ?? '' ) );
						$wp_edit_url     = esc_url_raw( (string) ( $fresh->edit_url ?? '' ) );
						$destination_host = (string) wp_parse_url( $wp_post_url ?: $destination_url, PHP_URL_HOST );
						$destination_host = (string) preg_replace( '/^www\./i', '', strtolower( $destination_host ) );
						if ( '' !== $destination_site_key || '' !== $destination_url || '' !== $wp_post_url ) {
							$destination_context = [
								'destination_site_key'  => $destination_site_key,
								'destination_site_name' => sanitize_text_field( $destination_name ),
								'destination_site_url'  => esc_url_raw( $destination_url ),
								'destination_site_host' => sanitize_text_field( $destination_host ),
								'wp_post_id'            => (int) ( $fresh->wp_post_id ?? 0 ),
								'wp_post_url'           => $wp_post_url,
								'wp_edit_url'           => $wp_edit_url,
								'publication_state'     => 'completed',
							];
						}

						Costudio_Telemetry::queue( 'production.completed', array_merge( [
							'production_id'     => $production_id,
							'production_uid'    => $production_uid,
							'type'              => Costudio_Article_Types::normalize( $inputs['article_type'] ?? '' ),
							'locale'            => function_exists( 'costudio_resolve_ui_locale_from_options' ) ? costudio_resolve_ui_locale_from_options() : 'pt_BR',
							'model'             => $totals->model ?? '',
							'grounding'         => ( (int) ( $totals->grounding_count ?? 0 ) > 0 ),
							'total_duration_ms' => (int) $totals->total_latency_ms,
							'total_cost_usd'    => (float) $totals->total_cost_usd,
							'total_tokens_in'   => (int) $totals->total_input_tokens,
							'total_tokens_out'  => (int) $totals->total_output_tokens,
							'agents_run'        => (int) $totals->agents_count,
							'image_count'       => $img_qty,
						], $destination_context ) );

						// Force immediate flush for completed productions.
						Costudio_Telemetry::flush();
					}
				}
			} catch ( Costudio_Upgrade_Required_Exception $e ) {
				Costudio_Production::mark_upgrade_required(
					$production_id,
					$e->get_error_code(),
					$e->getMessage(),
					$e->get_upgrade_url(),
					$agent_num
				);
				if ( class_exists( 'Costudio_BYOK_Gateway' ) ) {
					Costudio_BYOK_Gateway::mark_applied( $production_id, $agent_num );
				}
			} catch ( Costudio_Generation_Deferred_Exception $e ) {
				// Chunked remote polling: the AI job for this SAME agent is still
				// processing after one short chunk (~45s) — nothing failed and
				// nothing finished. Release the lock and re-dispatch this exact
				// agent (never $next_agent) in a fresh background request, which
				// resumes polling the same remote job instead of submitting a new
				// one. Deliberately bypasses the manual-approval gate and the
				// "advance to next agent" logic further up, since those only
				// apply once an agent has genuinely completed. Mirrors the
				// chunked image-publishing pattern (Costudio_Image_Deferred_Exception).
				self::release_lock( $production_id );

				$fresh = Costudio_Production::get_by_id( $production_id );
				if ( $fresh && in_array( $fresh->status, [ 'error', 'paused', 'cancelled', 'pending_upgrade' ], true ) ) {
					return;
				}
				if ( $fresh ) {
					$fresh->heartbeat = current_time( 'mysql', true );
					$fresh->save();
				}

				$is_durable_byok = 'byok_durable_job_active' === $e->getMessage();
				self::pipeline_log_append(
					$production_id,
					$is_durable_byok
						? __( 'The dedicated BYOK worker is processing this agent — waiting for its durable result…', 'contentosapp-studio' )
						: __( 'Job de IA ainda em processamento no servidor remoto — continuando em segundo plano…', 'contentosapp-studio' ),
					'info',
					$agent_num
				);

				$new_token = wp_generate_password( 32, false );
				self::prime_background_token( $production_id, $new_token, $is_durable_byok ? 'byok_readback' : 'chunk_resume' );
				if ( $is_durable_byok ) {
					self::schedule_background_check( $production_id, $new_token, 10 );
				} else {
					self::kick_off_background( $production_id, $new_token );
				}
				return;
			} catch ( Throwable $e ) {
				$fresh = Costudio_Production::get_by_id( $production_id );
				$error_message = $agent_num > $agent_qty
					? sprintf( 'Falha ao finalizar/publicar no WordPress: %s', $e->getMessage() )
					: $e->getMessage();
				if ( $fresh ) {
					$fresh->mark_failed( $agent_num, $error_message, $e );
				} else {
					Costudio_Production::set_error( $production_id, $error_message, $agent_num );
				}
				if ( class_exists( 'Costudio_BYOK_Gateway' ) ) {
					Costudio_BYOK_Gateway::mark_applied( $production_id, $agent_num );
				}
			}
		} finally {
			// Always release the lock, even on early returns or exceptions.
			self::release_lock( $production_id );
			if ( class_exists( 'Costudio_Production_Queue' ) ) {
				Costudio_Production_Queue::maybe_dispatch_next( 'runner_finally' );
			}
		}
	}

	// ── WP Hooks ─────────────────────────────────────────────────────────────

	/**
	 * REST route callback.
	 */
	public static function rest_run_background( WP_REST_Request $request ): void {
		$production_id = absint( $request->get_param( 'production_id' ) );
		$secret_token  = sanitize_text_field( (string) $request->get_param( 'secret_token' ) );
		self::run( $production_id, $secret_token );
		wp_die();
	}

	/**
	 * Cron hook callback.
	 */
	public static function cron_run_background( int $production_id, string $secret_token ): void {
		self::run( $production_id, $secret_token );
	}

	public static function register_rest_route(): void {
		register_rest_route( 'costudio/v1', '/run-background', [
			'methods'             => 'POST',
			'callback'            => [ self::class, 'rest_run_background' ],
			'permission_callback' => [ self::class, 'permission_check_run_background' ],
		] );
	}

	/**
	 * Permission callback for the background runner REST endpoint.
	 *
	 * The endpoint is intended to be hit only by the server's own loopback
	 * HTTP request (fired from kick_off_background). Allowing it on any IP
	 * with `__return_true` would let external actors flood the endpoint with
	 * tokenless requests and burn CPU on every miss — even though the actual
	 * token validation inside run() would still reject them.
	 *
	 * Strategy:
	 *   1. Allow if REMOTE_ADDR matches loopback or the server's own IP.
	 *   2. Otherwise, apply a per-IP rate limit (60 req/min) as defense in
	 *      depth — pipeline normal fires ~7 requests / 2-3 minutes, well
	 *      under the cap.
	 *
	 * The token check inside run() remains the authoritative guard.
	 */
	public static function permission_check_run_background( WP_REST_Request $request ) {
		$production_id = absint( $request->get_param( 'production_id' ) );
		$secret_token  = sanitize_text_field( (string) $request->get_param( 'secret_token' ) );

		if ( $production_id <= 0 || '' === $secret_token ) {
			return new WP_Error( 'costudio_background_missing_token', __( 'Missing background runner credentials.', 'contentosapp-studio' ), array( 'status' => 403 ) );
		}

		$stored = get_transient( "costudio_bg_{$production_id}" );
		if ( ! is_string( $stored ) || ! hash_equals( $stored, $secret_token ) ) {
			return new WP_Error( 'costudio_background_invalid_token', __( 'Invalid background runner token.', 'contentosapp-studio' ), array( 'status' => 403 ) );
		}

		$remote = isset( $_SERVER['REMOTE_ADDR'] ) ? sanitize_text_field( wp_unslash( $_SERVER['REMOTE_ADDR'] ) ) : '';

		if ( self::is_local_request( $remote ) ) {
			return true;
		}

		return self::throttle_rest_endpoint( $remote )
			? true
			: new WP_Error( 'costudio_background_rate_limited', __( 'Background runner rate limit reached.', 'contentosapp-studio' ), array( 'status' => 429 ) );
	}

	/**
	 * True if the request originates from the server itself.
	 */
	private static function is_local_request( string $remote_addr ): bool {
		if ( $remote_addr === '' ) {
			return false;
		}

		if ( in_array( $remote_addr, [ '127.0.0.1', '::1' ], true ) ) {
			return true;
		}

		// Some shared hosts route loopback through the server's public IP.
		$server_ip = isset( $_SERVER['SERVER_ADDR'] ) ? sanitize_text_field( wp_unslash( $_SERVER['SERVER_ADDR'] ) ) : '';
		if ( $server_ip !== '' && hash_equals( $server_ip, $remote_addr ) ) {
			return true;
		}

		return false;
	}

	/**
	 * Per-IP rate limit for the REST endpoint.
	 * 60 requests per minute is generous for legitimate use (a full pipeline
	 * fires ~7 requests across 2-3 minutes) and tight enough to neutralize
	 * casual flood attempts.
	 */
	private static function throttle_rest_endpoint( string $remote_addr ): bool {
		if ( $remote_addr === '' ) {
			$remote_addr = 'unknown';
		}

		$key   = 'costudio_rest_throttle_' . substr( md5( $remote_addr ), 0, 12 );
		$count = (int) get_transient( $key );

		if ( $count >= 60 ) {
			costudio_log_debug( 'runner.rest_rate_limit', [ 'remote_addr' => $remote_addr ] );
			return false;
		}

		set_transient( $key, $count + 1, MINUTE_IN_SECONDS );
		return true;
	}

	// ── Manual Control ───────────────────────────────────────────────────────

	/**
	 * Manually resume a paused or errored pipeline.
	 */
	public static function resume_pipeline_manual( int $production_id ) {
		$production = Costudio_Production::get_by_id( $production_id );
		if ( ! $production ) {
			return new WP_Error( 'not_found', 'Production not found.' );
		}

		$current_agent = max( 1, (int) ( $production->current_agent ?? 1 ) );
		if ( 'error' === (string) $production->status ) {
			// Resuming an errored agent is a new attempt/session. Paused jobs,
			// on the other hand, preserve their remote job and timing boundary.
			Costudio_Production::reset_for_retry( $production_id, $current_agent, 'pending' );
		} else {
			Costudio_Production::mark_pending( $production_id, $current_agent, true );
		}

		if ( class_exists( 'Costudio_Production_Queue' ) ) {
			return Costudio_Production_Queue::submit( $production_id );
		}

		Costudio_Production::mark_pipeline_started( $production_id, $current_agent );

		$token = wp_generate_password( 32, false );
		self::prime_background_token( $production_id, $token, 'manual_resume' );
		self::kick_off_background( $production_id, $token );

		return true;
	}

	// ── Metrics & Monitoring ────────────────────────────────────────────────

	public static function get_perf_metrics(): array {
		return [
			'queue_deferrals' => 0,
			'total_steps'     => 0,
			'failed_steps'    => 0,
			'avg_latency'     => 0,
		];
	}

	public static function get_slot_status(): array {
		global $wpdb;

		$table   = $wpdb->prefix . Costudio_Production::TABLE;
		$running = (int) $wpdb->get_var( "SELECT COUNT(*) FROM {$table} WHERE status = 'running'" );

		return [
			'total_slots' => 1,
			'used_slots'  => min( 1, $running ),
			'free_slots'  => $running > 0 ? 0 : 1,
		];
	}

	public static function schedule_step( int $production_id, int $agent_num ): void {
		$token = wp_generate_password( 32, false );
		self::prime_background_token( $production_id, $token, 'schedule_step' );
		wp_schedule_single_event( time(), self::CRON_HOOK, [ $production_id, $token ] );
	}
}
