<?php
/**
 * Costudio_Video_Discovery
 *
 * Item 3, subitem 3B (Fatia 2) — automatic YouTube video candidate discovery.
 * Runs only when the Strategist (Agent 02) has already concluded (its
 * VIDEO_OPPORTUNITY declaration is the trigger) and only when the user did
 * NOT already provide a working candidate via 3A (Costudio_Link_Reader /
 * Costudio_Pipeline_Links::find_video_candidate_urls()).
 *
 * Everything here is deliberately conservative: a single automatic attempt
 * per production, an atomic distributed lock (mirroring
 * Costudio_Runner::try_acquire_lock()), a persisted intermediate "processing"
 * state written BEFORE any network call, a hard time budget so a slow
 * upstream call can never hold the lock past its lease, and a relevance gate
 * that is independent of search-result position — a fast/irrelevant result
 * must never win just by ranking first.
 *
 * @package Contentosapp_Studio
 * @since   1.2.0 (Item 3B)
 */

defined( 'ABSPATH' ) || exit;

final class Costudio_Video_Discovery {

	/** Filter kept as an emergency/operational kill switch. */
	const FILTER_ENABLED = 'costudio_video_discovery_enabled';

	/** Native product default: automatic discovery is available without setup. */
	const DEFAULT_ENABLED = true;

	/** Total wall-clock budget for one discovery attempt (search + up to 3 oEmbeds). */
	const BUDGET_SECONDS = 90;

	/** Lock/lease safety margin ABOVE the budget — not the budget itself. */
	const LEASE_SECONDS = 120;

	/** Below this many seconds remaining, don't even start the provider search. */
	const MIN_BUDGET_FOR_SEARCH = 15;

	/** Below this many seconds remaining, don't attempt another oEmbed validation. */
	const MIN_BUDGET_FOR_OEMBED = 5;

	/** Hard cap on raw candidates considered from the provider's results. */
	const MAX_RAW_CANDIDATES = 5;

	/** Hard cap on oEmbed validation attempts among relevant candidates. */
	const MAX_OEMBED_ATTEMPTS = 3;

	/** Discard a token shorter than this (Unicode-safe length, no mbstring). */
	const MIN_TOKEN_CHARS = 3;

	/** Cap on QUERY_HINT length before it is ever used in a search query. */
	const QUERY_HINT_MAX_CHARS = 160;

	/**
	 * Stopwords + domain-generic terms (EN/PT/ES). Excluding "video"/"tutorial"/
	 * "youtube" and their translations is deliberate: without it, almost any
	 * YouTube search result "matches" the keyword purely by being a video.
	 */
	const STOPWORDS = array(
		// EN
		'the', 'and', 'for', 'with', 'how', 'what', 'why', 'from', 'your', 'you',
		'this', 'that', 'are', 'was', 'were', 'into', 'about', 'video', 'videos',
		'youtube', 'tutorial', 'guide', 'watch',
		// PT
		'como', 'para', 'com', 'uma', 'sobre', 'que', 'são', 'foi', 'vídeo',
		'vídeos', 'guia', 'assistir', 'por',
		// ES
		'cómo', 'con', 'una', 'qué', 'son', 'fue', 'vídeo', 'vídeos', 'video',
		'videos', 'guía', 'ver',
	);

	private function __construct() {
	}

	/**
	 * Represent a Throwable safely for logging — never the raw message, since
	 * it may carry a query string, URL, or provider parameter from the
	 * search/oEmbed call that triggered it (Codex review, 2026-08-20).
	 * Callers that need to correlate two log lines about the "same"
	 * exception can compare message_hash values without ever seeing the
	 * original text.
	 *
	 * @return array{error_class:string,message_hash:string}
	 */
	public static function sanitize_exception_for_log( \Throwable $e ): array {
		return array(
			'error_class'  => get_class( $e ),
			'message_hash' => substr( hash( 'sha256', $e->getMessage() ), 0, 12 ),
		);
	}

	// ── Public entry point ────────────────────────────────────────────────

	/**
	 * Attempt automatic video discovery for a production, exactly once.
	 * Safe to call repeatedly — every call after the first (successful or
	 * not) short-circuits on `video_discovery.attempted`.
	 *
	 * @param int    $production_id
	 * @param object $production     Row with ->output_02, ->links_internos.
	 * @param array  $inputs         Current production inputs (already includes
	 *                                'video_candidate' from the 3A resolution).
	 * @return array The Costudio_Link_Reader::read_video()-shaped candidate
	 *               (ok=true when selected), or ['ok'=>false,...] otherwise.
	 */
	public static function attempt( int $production_id, object $production, array $inputs ): array {
		$existing_discovery = is_array( $inputs['video_discovery'] ?? null ) ? $inputs['video_discovery'] : array();

		// Already attempted (any terminal OR still-processing-but-not-expired) →
		// never re-run. This is the ONLY idempotency gate for 3B — never
		// video_candidate, which 3A already writes independently (including ok=false).
		if ( ! empty( $existing_discovery['attempted'] ) ) {
			if ( 'processing' === ( $existing_discovery['status'] ?? '' ) ) {
				$lease_expires = strtotime( (string) ( $existing_discovery['lease_expires_at'] ?? '' ) ) ?: 0;
				if ( $lease_expires > 0 && time() < $lease_expires ) {
					// Another attempt is genuinely still in flight — do nothing this run.
					return array( 'ok' => false, 'error' => 'discovery_in_progress' );
				}
				// Abandoned attempt (crashed before writing a terminal state) —
				// terminalize as interrupted, WITHOUT starting a new attempt.
				self::finalize_abandoned( $production_id, $existing_discovery );
				return array( 'ok' => false, 'error' => 'discovery_interrupted' );
			}
			return array( 'ok' => false, 'error' => 'discovery_already_attempted' );
		}

		if ( ! apply_filters( self::FILTER_ENABLED, self::DEFAULT_ENABLED ) ) {
			return array( 'ok' => false, 'error' => 'discovery_disabled' );
		}

		$opportunity = self::parse_video_opportunity( (string) ( $production->output_02 ?? '' ) );
		if ( empty( $opportunity['helps'] ) ) {
			self::persist_terminal( $production_id, $inputs, 'no_candidate', '', 'video_opportunity_no' );
			return array( 'ok' => false, 'error' => 'video_opportunity_no' );
		}

		if ( ! class_exists( 'Costudio_Pipeline' ) ) {
			return array( 'ok' => false, 'error' => 'pipeline_class_missing' );
		}
		$policy = Costudio_Pipeline::search_policy_for( $inputs );
		if ( empty( $policy['enabled'] ) ) {
			self::persist_terminal( $production_id, $inputs, 'no_candidate', '', 'search_disabled' );
			return array( 'ok' => false, 'error' => 'search_disabled' );
		}
		if ( 'external' === $policy['mode'] && ( ! class_exists( 'Costudio_Search_Engine' ) || ! Costudio_Search_Engine::provider_is_structured( $policy['provider'] ) ) ) {
			// Gemini (or any future non-structured provider) degrades silently —
			// this is correct behaviour, not an error condition.
			self::persist_terminal( $production_id, $inputs, 'no_candidate', '', 'provider_not_structured' );
			return array( 'ok' => false, 'error' => 'provider_not_structured' );
		}

		// ── Atomic lock ──────────────────────────────────────────────────
		$owner_token = wp_generate_password( 20, false );
		$lock_value  = self::try_acquire_lock( $production_id, $owner_token );
		if ( '' === $lock_value ) {
			self::log( 'warn', 'lock_contention', array( 'production_id' => $production_id ) );
			return array( 'ok' => false, 'error' => 'lock_contention' );
		}

		try {
			// Re-read inputs from inside the lock — avoid acting on data that
			// went stale while we were waiting to acquire it.
			$fresh_inputs = self::reread_inputs( $production_id, $inputs );
			if ( ! empty( ( $fresh_inputs['video_discovery'] ?? array() )['attempted'] ?? false ) ) {
				// Someone else finished (or started) between our first check and
				// acquiring the lock — back off, nothing to do.
				self::release_lock( $production_id, $lock_value );
				return array( 'ok' => false, 'error' => 'discovery_already_attempted' );
			}

			$attempt_id = $owner_token;
			$started_at = time();
			$lease_at   = $started_at + self::LEASE_SECONDS;

			if ( ! self::persist_processing( $production_id, $fresh_inputs, $attempt_id, $started_at, $lease_at ) ) {
				// The intermediate write wasn't confirmed by the database — do
				// NOT start any network call under an unconfirmed state (Codex
				// review, 2026-08-20). Nothing was durably marked "attempted",
				// so a future call is free to retry this production normally.
				self::log( 'warn', 'processing_persist_failed', array( 'production_id' => $production_id ) );
				return array( 'ok' => false, 'error' => 'processing_persist_failed' );
			}
			self::log( 'info', 'processing_started', array(
				'production_id' => $production_id,
				'mode'          => $policy['mode'],
				'provider'      => $policy['provider'],
			) );

			$budget_start = microtime( true );
			$deadline     = $budget_start + self::BUDGET_SECONDS;
			$remaining    = static function () use ( $budget_start ): float {
				return max( 0.0, self::BUDGET_SECONDS - ( microtime( true ) - $budget_start ) );
			};

			$result = array( 'ok' => false, 'error' => 'no_candidate' );
			$status = 'no_candidate';
			$reason = 'no_relevant_candidate';

			if ( $remaining() >= self::MIN_BUDGET_FOR_SEARCH ) {
				$raw_candidates = self::search_video_candidates(
					(string) ( $inputs['keyword'] ?? '' ),
					$opportunity['query_hint'],
					$policy,
					$deadline
				);

				$relevant = self::filter_relevant(
					(string) ( $inputs['keyword'] ?? '' ),
					$opportunity['query_hint'],
					$raw_candidates
				);

				self::log( 'info', 'candidates_scored', array(
					'production_id'   => $production_id,
					'raw_count'       => count( $raw_candidates ),
					'relevant_count'  => count( $relevant ),
				) );

				$attempts = 0;
				foreach ( $relevant as $candidate ) {
					if ( $attempts >= self::MAX_OEMBED_ATTEMPTS ) {
						break;
					}
					if ( $remaining() < self::MIN_BUDGET_FOR_OEMBED ) {
						break;
					}
					$attempts++;
					$oembed_timeout = max( 1, (int) round( min( $remaining(), 6.0 ) ) );
					$confirmed      = class_exists( 'Costudio_Link_Reader' )
						? Costudio_Link_Reader::read_video( $candidate['url'], array( 'timeout' => $oembed_timeout ) )
						: array( 'ok' => false );

					if ( ! empty( $confirmed['ok'] ) ) {
						$result = $confirmed;
						$status = 'selected';
						$reason = '';
						break;
					}
				}
			} else {
				$reason = 'insufficient_budget';
			}

			// ── Conditional final write — reconfirm ownership before writing. ──
			if ( self::reconfirm_and_finalize( $production_id, $lock_value, $attempt_id, $status, $result, $reason ) ) {
				self::log( 'info', $status, array( 'production_id' => $production_id, 'reason' => $reason ) );
			} else {
				// Lost the lock (a later process already terminalized this
				// attempt as abandoned) — our result is stale, discard it silently.
				self::log( 'warn', 'stale_result_discarded', array( 'production_id' => $production_id ) );
				return array( 'ok' => false, 'error' => 'stale_result_discarded' );
			}

			return $result;
		} finally {
			self::release_lock( $production_id, $lock_value );
		}
	}

	// ── VIDEO_OPPORTUNITY parsing ────────────────────────────────────────

	/**
	 * @return array{helps:bool,section_hint:string,query_hint:string}
	 */
	public static function parse_video_opportunity( string $output_02 ): array {
		$empty = array( 'helps' => false, 'section_hint' => '', 'query_hint' => '' );

		if ( ! preg_match(
			'/###\s*VIDEO_OPPORTUNITY\s*\n\s*HELPS:\s*([^\n]*)\n\s*SECTION_HINT:\s*([^\n]*)\n\s*QUERY_HINT:\s*([^\n]*)/i',
			$output_02,
			$m
		) ) {
			return $empty;
		}

		$helps = 'YES' === strtoupper( trim( $m[1] ) );
		if ( ! $helps ) {
			return $empty;
		}

		$section_hint = trim( $m[2] );
		$query_hint   = trim( $m[3] );
		if ( 'NONE' === strtoupper( $section_hint ) ) {
			$section_hint = '';
		}
		if ( 'NONE' === strtoupper( $query_hint ) || '' === $query_hint ) {
			// HELPS=YES but no usable hint — nothing to search with.
			return $empty;
		}

		return array(
			'helps'        => true,
			'section_hint' => $section_hint,
			'query_hint'   => self::sanitize_query_hint( $query_hint ),
		);
	}

	/**
	 * Defensive truncation — never trust the model to honour "one line, ~160
	 * chars" on its own (same distrust already applied elsewhere in the pipeline).
	 */
	private static function sanitize_query_hint( string $hint ): string {
		$first_line = preg_split( '/\r\n|\r|\n/', $hint )[0] ?? '';
		$first_line = trim( $first_line );
		if ( 1 === preg_match( '/^(.{1,' . self::QUERY_HINT_MAX_CHARS . '})/us', $first_line, $m ) ) {
			return trim( $m[1] );
		}
		return $first_line;
	}

	// ── Relevance (Unicode-safe tokenization, no mbstring) ────────────────

	/**
	 * @return array<int,string>
	 */
	public static function significant_tokens( string $text ): array {
		$tokens = preg_split( '/[^\p{L}\p{N}]+/u', $text, -1, PREG_SPLIT_NO_EMPTY ) ?: array();
		$out    = array();
		foreach ( $tokens as $t ) {
			if ( 1 !== preg_match( '/^.{' . self::MIN_TOKEN_CHARS . ',}$/us', $t ) ) {
				continue;
			}
			if ( self::is_stopword( $t ) ) {
				continue;
			}
			$out[] = $t;
		}
		return $out;
	}

	private static function is_stopword( string $token ): bool {
		foreach ( self::STOPWORDS as $stopword ) {
			if ( 1 === preg_match( '/^' . preg_quote( $stopword, '/' ) . '$/iu', $token ) ) {
				return true;
			}
		}
		return false;
	}

	/** Count of $needles present (case-insensitive, Unicode-safe) in $haystack. */
	private static function tokens_matched( array $haystack, array $needles ): int {
		$count = 0;
		foreach ( $needles as $needle ) {
			foreach ( $haystack as $hay ) {
				if ( 1 === preg_match( '/^' . preg_quote( $needle, '/' ) . '$/iu', $hay ) ) {
					$count++;
					break;
				}
			}
		}
		return $count;
	}

	/**
	 * Binary relevance gate, independent of provider position. Keyword tokens
	 * are mandatory (must appear in the TITLE); hint tokens can only reinforce
	 * a second match when the keyword itself has 2+ significant tokens — the
	 * hint alone can never approve a candidate.
	 */
	public static function is_relevant( string $keyword, string $hint, string $title, string $snippet ): bool {
		$keyword_tokens = self::significant_tokens( $keyword );
		if ( empty( $keyword_tokens ) ) {
			return false; // nothing significant to judge relevance against
		}

		$title_tokens = self::significant_tokens( $title );
		$keyword_in_title = self::tokens_matched( $title_tokens, $keyword_tokens ) >= 1;
		if ( ! $keyword_in_title ) {
			return false;
		}

		if ( count( $keyword_tokens ) < 2 ) {
			return true; // single-term keyword — the title match above is enough
		}

		$hint_tokens = self::significant_tokens( $hint );
		$combined    = array_values( array_unique( array_merge( $keyword_tokens, $hint_tokens ) ) );
		$all_tokens  = self::significant_tokens( $title . ' ' . $snippet );
		return self::tokens_matched( $all_tokens, $combined ) >= 2;
	}

	/**
	 * @param array<int,array{url:string,title:string,snippet:string}> $candidates
	 * @return array<int,array{url:string,title:string,snippet:string}> Only the
	 *         relevant ones, in their original (provider) order.
	 */
	public static function filter_relevant( string $keyword, string $hint, array $candidates ): array {
		$out = array();
		foreach ( $candidates as $candidate ) {
			if ( self::is_relevant( $keyword, $hint, (string) ( $candidate['title'] ?? '' ), (string) ( $candidate['snippet'] ?? '' ) ) ) {
				$out[] = $candidate;
			}
		}
		return $out;
	}

	// ── Search orchestration (mirrors Costudio_Pipeline's native/external routing) ──

	/**
	 * @param float $deadline Absolute microtime(true) budget for this call. The
	 *              native engine re-derives its own per-transport timeout from
	 *              this deadline and re-checks it before every attempt/fallback/
	 *              retry internally (Costudio_Search_Engine_ContentOS::search()),
	 *              so a slow upstream can never silently blow past the caller's
	 *              (Costudio_Video_Discovery::attempt()) 90s/120s budget.
	 * @return array<int,array{url:string,title:string,snippet:string}>
	 */
	public static function search_video_candidates( string $keyword, string $hint, array $policy, float $deadline ): array {
		$query = trim( $keyword . ( '' !== $hint ? ' ' . $hint : '' ) );
		if ( '' === $query ) {
			return array();
		}

		$timeout = (int) max( 1, min( 60, floor( $deadline - microtime( true ) ) ) );

		$sources = array();
		if ( 'native' === $policy['mode'] ) {
			if ( ! class_exists( 'Costudio_Search_Engine_ContentOS' ) ) {
				return array();
			}
			$engine = new Costudio_Search_Engine_ContentOS();
			$result = $engine->search( $query, array( 'timeout' => $timeout, 'deadline' => $deadline ) );
			$sources = is_array( $result['sources'] ?? null ) ? $result['sources'] : array();
		} elseif ( 'external' === $policy['mode'] ) {
			if ( ! class_exists( 'Costudio_Search_Engine' ) ) {
				return array();
			}
			// BYOK providers (Tavily/Serper/Exa) make a single HTTP call each —
			// no internal retry/fallback loop — so a plain per-call timeout is
			// already sufficient; no $deadline propagation needed here.
			$engine  = new Costudio_Search_Engine();
			$result  = $engine->search_structured( $query . ' site:youtube.com', array( 'timeout' => $timeout ) );
			$sources = is_array( $result['sources'] ?? null ) ? $result['sources'] : array();
		} else {
			return array();
		}

		$candidates = array();
		foreach ( $sources as $src ) {
			$url = trim( (string) ( $src['url'] ?? '' ) );
			if ( '' === $url || ! class_exists( 'Costudio_Link_Reader' ) || ! Costudio_Link_Reader::is_youtube_url( $url ) ) {
				continue;
			}
			$candidates[] = array(
				'url'     => $url,
				'title'   => (string) ( $src['title'] ?? '' ),
				'snippet' => (string) ( $src['snippet'] ?? '' ),
			);
			if ( count( $candidates ) >= self::MAX_RAW_CANDIDATES ) {
				break;
			}
		}
		return $candidates;
	}

	// ── Atomic lock (mirrors Costudio_Runner::try_acquire_lock() exactly) ──

	private static function lock_key( int $production_id ): string {
		return "costudio_video_discovery_lock_{$production_id}";
	}

	/**
	 * @return string The acquired lock value ("owner_token|timestamp"), or ''
	 *                if another process holds an active lock.
	 */
	private static function try_acquire_lock( int $production_id, string $owner_token ): string {
		global $wpdb;

		$lock_key   = self::lock_key( $production_id );
		$lock_value = $owner_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 ( 1 === (int) $inserted ) {
			wp_cache_delete( $lock_key, 'options' );
			return $lock_value;
		}

		$existing = $wpdb->get_var( $wpdb->prepare(
			"SELECT option_value FROM {$wpdb->options} WHERE option_name = %s",
			$lock_key
		) );
		if ( ! $existing || false === strpos( $existing, '|' ) ) {
			return '';
		}

		$parts       = explode( '|', $existing, 2 );
		$existing_ts = (int) ( $parts[1] ?? 0 );
		if ( $existing_ts <= 0 || ( time() - $existing_ts ) <= self::LEASE_SECONDS ) {
			return ''; // active lock — back off
		}

		// Stale — take it over only if nobody changed it meanwhile (CAS).
		$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 ? $lock_value : '';
	}

	/**
	 * Reconfirm ownership via the lock's CAS, and only if that succeeds, write
	 * the terminal video_discovery/video_candidate state.
	 *
	 * The ownership check MUST rewrite the lock to a genuinely NEW value, not
	 * the same value it already holds. On real MySQL/MariaDB (the default,
	 * without CLIENT_FOUND_ROWS), an UPDATE whose SET value equals the current
	 * stored value reports 0 affected rows even though the WHERE matched — so
	 * a same-value CAS would misreport "lock lost" on every normal run and
	 * strand video_discovery in "processing" forever (Codex review, 2026-08-20;
	 * the original implementation had this bug and it was masked by a fake
	 * in-memory $wpdb that didn't reproduce the real no-op-UPDATE behaviour).
	 *
	 * This is also what makes a late/straggler result harmless: if another
	 * process already stole the lock (because our lease looked expired to
	 * it), the CAS below affects 0 rows and we know not to write anything.
	 */
	private static function reconfirm_and_finalize(
		int $production_id,
		string $lock_value,
		string $attempt_id,
		string $status,
		array $result,
		string $reason
	): bool {
		global $wpdb;

		$lock_key       = self::lock_key( $production_id );
		$new_lock_value = $lock_value . '~fin~' . wp_generate_password( 8, false );

		$still_own = $wpdb->query( $wpdb->prepare(
			"UPDATE {$wpdb->options} SET option_value = %s WHERE option_name = %s AND option_value = %s",
			$new_lock_value,
			$lock_key,
			$lock_value
		) );
		wp_cache_delete( $lock_key, 'options' );

		if ( 1 !== (int) $still_own ) {
			return false;
		}

		$success = false;
		try {
			$inputs = self::reread_inputs( $production_id, array() );

			// Second line of defence: the persisted attempt_id must still be ours.
			$current = is_array( $inputs['video_discovery'] ?? null ) ? $inputs['video_discovery'] : array();
			if ( ( $current['attempt_id'] ?? null ) !== $attempt_id || 'processing' !== ( $current['status'] ?? '' ) ) {
				return false;
			}

			$inputs['video_discovery'] = array(
				'attempted'    => true,
				'status'       => $status,
				'attempt_id'   => $attempt_id,
				'reason'       => $reason,
				'attempted_at' => gmdate( 'c' ),
			);
			if ( 'selected' === $status && ! empty( $result['ok'] ) ) {
				$candidate           = $result;
				$candidate['origin'] = 'auto_discovery';
				$inputs['video_candidate'] = $candidate;
			}

			// Never claim the terminal state was saved when the database
			// rejected the write (Codex review, 2026-08-20).
			$success = class_exists( 'Costudio_Production' )
				? Costudio_Production::update_inputs( $production_id, $inputs )
				: false;

			return $success;
		} finally {
			// We own $new_lock_value at this point regardless of $success —
			// release it now rather than waiting for the caller's finally
			// block (which still holds the OLD $lock_value and will no-op).
			$wpdb->query( $wpdb->prepare(
				"DELETE FROM {$wpdb->options} WHERE option_name = %s AND option_value = %s",
				$lock_key,
				$new_lock_value
			) );
			wp_cache_delete( $lock_key, 'options' );
		}
	}

	/** Terminalize without ever having acquired the lock (early-exit paths). @return bool Write confirmed. */
	private static function persist_terminal( int $production_id, array $inputs, string $status, string $attempt_id, string $reason ): bool {
		$inputs['video_discovery'] = array(
			'attempted'    => true,
			'status'       => $status,
			'attempt_id'   => $attempt_id,
			'reason'       => $reason,
			'attempted_at' => gmdate( 'c' ),
		);
		return class_exists( 'Costudio_Production' )
			? Costudio_Production::update_inputs( $production_id, $inputs )
			: false;
	}

	/** @return bool Whether the intermediate "processing" state was durably confirmed by the database. */
	private static function persist_processing( int $production_id, array $inputs, string $attempt_id, int $started_at, int $lease_at ): bool {
		$inputs['video_discovery'] = array(
			'attempted'        => true,
			'status'           => 'processing',
			'attempt_id'       => $attempt_id,
			'started_at'       => gmdate( 'c', $started_at ),
			'lease_expires_at' => gmdate( 'c', $lease_at ),
		);
		return class_exists( 'Costudio_Production' )
			? Costudio_Production::update_inputs( $production_id, $inputs )
			: false;
	}

	/**
	 * An abandoned `processing` attempt is terminalized as interrupted — never
	 * retried automatically.
	 *
	 * Must take the (presumably stale) lock atomically via the SAME CAS
	 * try_acquire_lock() uses before writing anything. Without this, a
	 * process that only *looked* abandoned from a possibly-stale `inputs`
	 * snapshot could overwrite a correct terminal state written moments
	 * later by the real owner finishing late — the exact "process A wakes
	 * up late" race reconfirm_and_finalize() already guards against on the
	 * happy path, which this early-return path was missing entirely (Codex
	 * review, 2026-08-20). If the lock is still actively held, that means
	 * the real owner is alive (its lease_expires_at just hasn't been
	 * re-read here) — back off and do nothing.
	 */
	private static function finalize_abandoned( int $production_id, array $existing_discovery ): void {
		$owner_token = wp_generate_password( 20, false );
		$lock_value  = self::try_acquire_lock( $production_id, $owner_token );
		if ( '' === $lock_value ) {
			// Either the real owner is still actively holding the lock, or
			// another process already took over recovery — do nothing.
			return;
		}

		try {
			$inputs  = self::reread_inputs( $production_id, array() );
			$current = is_array( $inputs['video_discovery'] ?? null ) ? $inputs['video_discovery'] : array();
			// Only finalize if it's still the SAME abandoned attempt we
			// inspected — another process may have already handled it
			// between our read and now acquiring the lock.
			if ( 'processing' !== ( $current['status'] ?? '' ) || ( $current['attempt_id'] ?? null ) !== ( $existing_discovery['attempt_id'] ?? null ) ) {
				return;
			}
			$inputs['video_discovery'] = array(
				'attempted'    => true,
				'status'       => 'error',
				'attempt_id'   => (string) ( $existing_discovery['attempt_id'] ?? '' ),
				'reason'       => 'interrupted',
				'attempted_at' => gmdate( 'c' ),
			);
			if ( class_exists( 'Costudio_Production' ) ) {
				Costudio_Production::update_inputs( $production_id, $inputs );
			}
		} finally {
			self::release_lock( $production_id, $lock_value );
		}
	}

	private static function release_lock( int $production_id, string $lock_value ): void {
		if ( '' === $lock_value ) {
			return;
		}
		global $wpdb;
		$wpdb->query( $wpdb->prepare(
			"DELETE FROM {$wpdb->options} WHERE option_name = %s AND option_value = %s",
			self::lock_key( $production_id ),
			$lock_value
		) );
		wp_cache_delete( self::lock_key( $production_id ), 'options' );
	}

	/** Fresh read of inputs from the DB — never trust a possibly-stale copy. */
	private static function reread_inputs( int $production_id, array $fallback ): array {
		if ( ! class_exists( 'Costudio_Production' ) ) {
			return $fallback;
		}
		$fresh = Costudio_Production::get_by_id( $production_id );
		if ( ! $fresh ) {
			return $fallback;
		}
		$decoded = json_decode( (string) ( $fresh->inputs ?? '' ), true );
		return is_array( $decoded ) ? $decoded : $fallback;
	}

	// ── Logging — minimal, no QUERY_HINT/title/channel (Codex round 3) ────

	private static function log( string $level, string $event, array $context ): void {
		if ( ! function_exists( 'costudio_log_debug' ) && ! class_exists( 'Costudio_Logger' ) ) {
			return;
		}
		$message = 'pipeline.video_discovery.' . $event;
		if ( class_exists( 'Costudio_Logger' ) ) {
			if ( 'warn' === $level && method_exists( 'Costudio_Logger', 'warn' ) ) {
				Costudio_Logger::warn( $message, $context );
				return;
			}
			if ( method_exists( 'Costudio_Logger', 'info' ) ) {
				Costudio_Logger::info( $message, $context );
				return;
			}
		}
		if ( function_exists( 'costudio_log_debug' ) ) {
			costudio_log_debug( $message, $context );
		}
	}
}
