<?php
/**
 * Durable local job ledger for BYOK production calls.
 *
 * Phase 1 deliberately does not route provider traffic through this class yet.
 * It establishes the durable state machine, idempotency contract and operational
 * queries that the provider worker will use in the next phase.
 *
 * @package Contentosapp_Studio
 */

defined( 'ABSPATH' ) || exit;
// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.PreparedSQL.NotPrepared,WordPress.DB.DirectDatabaseQuery,PluginCheck.Security.DirectDB.UnescapedDBParameter -- Queries target a plugin-owned table; identifiers are fixed by class constants.

class Costudio_BYOK_Jobs {

	public const TABLE = 'costudio_byok_jobs';

	public const QUEUE_VERSION = 1;

	public const MODE_LEGACY           = 'legacy';
	public const MODE_DURABLE_LOOPBACK = 'durable_loopback';
	public const MODE_EXTERNAL_CLI     = 'external_cli';

	private const MODE_OPTION      = 'costudio_byok_execution_mode';
	private const CANARY_OPTION    = 'costudio_byok_canary_license_hashes';
	private const WORKER_OPTION    = 'costudio_byok_worker_heartbeat';
	private const CLAIM_LOCK_NAME  = 'costudio_byok_claim_gate';
	private const WORKER_LOCK_NAME = 'costudio_byok_worker_singleton';
	private const TERMINAL_STATES  = array( 'completed', 'failed', 'cancelled', 'uncertain', 'superseded' );
	private const OPEN_STATES      = array( 'queued', 'processing', 'cancel_requested' );
	private const ALLOWED_STATES   = array( 'queued', 'processing', 'completed', 'failed', 'cancel_requested', 'cancelled', 'uncertain', 'superseded' );

	/**
	 * Create or upgrade the durable BYOK jobs table.
	 */
	public static function install(): void {
		global $wpdb;

		$table   = self::table_name();
		$charset = $wpdb->get_charset_collate();
		require_once ABSPATH . 'wp-admin/includes/upgrade.php';

		dbDelta( "CREATE TABLE {$table} (
			id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
			job_id char(36) NOT NULL,
			production_id bigint(20) unsigned NOT NULL,
			production_uid varchar(100) NOT NULL DEFAULT '',
			agent_id tinyint unsigned NOT NULL,
			attempt_id char(36) NOT NULL,
			idempotency_key char(64) NOT NULL,
			request_hash char(64) NOT NULL,
			provider varchar(20) NOT NULL,
			model varchar(191) NOT NULL DEFAULT '',
			execution_profile_hash char(64) NOT NULL DEFAULT '',
			credential_fingerprint char(64) NOT NULL DEFAULT '',
			driver varchar(24) NOT NULL DEFAULT 'legacy',
			status varchar(24) NOT NULL DEFAULT 'queued',
			available_at datetime NOT NULL,
			worker_id varchar(191) NOT NULL DEFAULT '',
			claim_count int unsigned NOT NULL DEFAULT 0,
			claimed_at datetime DEFAULT NULL,
			processing_started_at datetime DEFAULT NULL,
			heartbeat_at datetime DEFAULT NULL,
			lease_expires_at datetime DEFAULT NULL,
			provider_dispatched_at datetime DEFAULT NULL,
			first_byte_at datetime DEFAULT NULL,
			response_received_at datetime DEFAULT NULL,
			request_ciphertext longtext DEFAULT NULL,
			partial_ciphertext longtext DEFAULT NULL,
			result_ciphertext longtext DEFAULT NULL,
			usage_json text DEFAULT NULL,
			error_code varchar(100) NOT NULL DEFAULT '',
			error_public text DEFAULT NULL,
			retry_safe tinyint(1) NOT NULL DEFAULT 0,
			terminal_source varchar(40) NOT NULL DEFAULT '',
			applied_at datetime DEFAULT NULL,
			usage_recorded_at datetime DEFAULT NULL,
			superseded_at datetime DEFAULT NULL,
			finished_at datetime DEFAULT NULL,
			payload_purged_at datetime DEFAULT NULL,
			queue_version tinyint unsigned NOT NULL DEFAULT 1,
			created_at datetime NOT NULL,
			updated_at datetime NOT NULL,
			PRIMARY KEY  (id),
			UNIQUE KEY job_id (job_id),
			UNIQUE KEY attempt_id (attempt_id),
			UNIQUE KEY idempotency_key (idempotency_key),
			KEY queue_claim (status, available_at, id),
			KEY production_attempt (production_id, agent_id, status, id),
			KEY lease_recovery (status, lease_expires_at),
			KEY cleanup_terminal (finished_at, payload_purged_at),
			KEY updated_at (updated_at)
		) ENGINE=InnoDB {$charset};" );
	}

	/**
	 * Resolve the current execution mode. A constant is an explicit operational
	 * override; existing installations remain on the legacy path by default.
	 */
	public static function mode(): string {
		$value = defined( 'COSTUDIO_BYOK_EXECUTION_MODE' )
			? (string) COSTUDIO_BYOK_EXECUTION_MODE
			: (string) get_option( self::MODE_OPTION, self::MODE_LEGACY );

		return self::normalize_mode( $value );
	}

	public static function set_mode( string $mode ): bool {
		$mode = self::normalize_mode( $mode );
		if ( defined( 'COSTUDIO_BYOK_EXECUTION_MODE' ) ) {
			return false;
		}
		if ( self::MODE_LEGACY !== $mode && ! self::pipeline_connected() ) {
			return false;
		}

		return update_option( self::MODE_OPTION, $mode, false ) || get_option( self::MODE_OPTION ) === $mode;
	}

	/**
	 * Phase gate. Phase 3 connects production submission and result application,
	 * while the legacy mode and the license canary remain fail-safe defaults.
	 */
	public static function pipeline_connected(): bool {
		return true;
	}

	/** Route new BYOK calls only after both operational gates are enabled. */
	public static function routes_new_jobs(): bool {
		return self::pipeline_connected()
			&& self::MODE_EXTERNAL_CLI === self::mode()
			&& self::current_license_is_canary();
	}

	public static function current_license_is_canary(): bool {
		if ( defined( 'COSTUDIO_BYOK_CANARY_ALL' ) && COSTUDIO_BYOK_CANARY_ALL ) {
			return true;
		}
		$hash = self::current_license_hash();
		return '' !== $hash && in_array( $hash, self::canary_hashes(), true );
	}

	public static function current_license_hash(): string {
		$key = trim( (string) get_option( 'costudio_license_key', '' ) );
		return '' === $key ? '' : hash( 'sha256', $key );
	}

	public static function canary_hashes(): array {
		$stored = get_option( self::CANARY_OPTION, array() );
		$stored = is_array( $stored ) ? $stored : array();
		return array_values( array_unique( array_filter( array_map( static function ( $hash ): string {
			$hash = strtolower( trim( (string) $hash ) );
			return 1 === preg_match( '/^[a-f0-9]{64}$/', $hash ) ? $hash : '';
		}, $stored ) ) ) );
	}

	public static function add_canary_hash( string $hash ): bool {
		$hash = strtolower( trim( $hash ) );
		if ( 1 !== preg_match( '/^[a-f0-9]{64}$/', $hash ) ) {
			return false;
		}
		$hashes = self::canary_hashes();
		if ( ! in_array( $hash, $hashes, true ) ) {
			$hashes[] = $hash;
		}
		return update_option( self::CANARY_OPTION, $hashes, false ) || self::canary_hashes() === $hashes;
	}

	public static function remove_canary_hash( string $hash ): bool {
		$hash   = strtolower( trim( $hash ) );
		$hashes = array_values( array_diff( self::canary_hashes(), array( $hash ) ) );
		return update_option( self::CANARY_OPTION, $hashes, false ) || self::canary_hashes() === $hashes;
	}

	public static function normalize_mode( string $mode ): string {
		$mode = sanitize_key( $mode );
		return in_array( $mode, self::allowed_modes(), true ) ? $mode : self::MODE_LEGACY;
	}

	public static function allowed_modes(): array {
		return array( self::MODE_LEGACY, self::MODE_DURABLE_LOOPBACK, self::MODE_EXTERNAL_CLI );
	}

	/**
	 * Idempotently create a queued job.
	 *
	 * Replaying the same idempotency key and request hash returns the original
	 * row. Reusing a key with different content is rejected before any provider
	 * execution can occur.
	 *
	 * @throws RuntimeException On an idempotency conflict or storage failure.
	 */
	public static function create( array $data ): array {
		global $wpdb;

		$production_id = (int) ( $data['production_id'] ?? 0 );
		$agent_id      = (int) ( $data['agent_id'] ?? 0 );
		if ( $production_id < 1 ) {
			throw new RuntimeException( 'byok_job_invalid_production_id' );
		}
		if ( $agent_id < 1 || $agent_id > 7 ) {
			throw new RuntimeException( 'byok_job_invalid_agent_id' );
		}

		$idempotency_key = self::hash_field( (string) ( $data['idempotency_key'] ?? '' ), 'idempotency_key' );
		$request_hash    = self::hash_field( (string) ( $data['request_hash'] ?? '' ), 'request_hash' );
		$existing        = self::get_by_idempotency_key( $idempotency_key );
		if ( $existing ) {
			if ( ! hash_equals( (string) $existing['request_hash'], $request_hash ) ) {
				throw new RuntimeException( 'byok_job_idempotency_conflict' );
			}
			$existing['created_now'] = false;
			return $existing;
		}

		$now        = current_time( 'mysql', true );
		$job_id     = self::uuid_field( (string) ( $data['job_id'] ?? wp_generate_uuid4() ), 'job_id' );
		$attempt_id = self::uuid_field( (string) ( $data['attempt_id'] ?? wp_generate_uuid4() ), 'attempt_id' );
		$provider   = sanitize_key( (string) ( $data['provider'] ?? '' ) );
		if ( ! in_array( $provider, array( 'claude', 'gemini', 'openai' ), true ) ) {
			throw new RuntimeException( 'byok_job_invalid_provider' );
		}
		$driver = self::normalize_mode( (string) ( $data['driver'] ?? self::mode() ) );
		$request_ciphertext = (string) ( $data['request_ciphertext'] ?? '' );
		if ( self::MODE_LEGACY !== $driver && 0 !== strpos( $request_ciphertext, 'FOSc2:' ) ) {
			throw new RuntimeException( 'byok_job_encrypted_payload_required' );
		}

		$row = array(
			'job_id'                 => $job_id,
			'production_id'          => $production_id,
			'production_uid'         => sanitize_text_field( (string) ( $data['production_uid'] ?? '' ) ),
			'agent_id'               => $agent_id,
			'attempt_id'             => $attempt_id,
			'idempotency_key'        => $idempotency_key,
			'request_hash'           => $request_hash,
			'provider'               => $provider,
			'model'                  => sanitize_text_field( (string) ( $data['model'] ?? '' ) ),
			'execution_profile_hash' => self::optional_hash( (string) ( $data['execution_profile_hash'] ?? '' ) ),
			'credential_fingerprint' => self::optional_hash( (string) ( $data['credential_fingerprint'] ?? '' ) ),
			'driver'                 => $driver,
			'status'                 => 'queued',
			'available_at'           => self::normalize_datetime( (string) ( $data['available_at'] ?? '' ), $now ),
			'request_ciphertext'     => $request_ciphertext,
			'queue_version'          => self::QUEUE_VERSION,
			'created_at'             => $now,
			'updated_at'             => $now,
		);

		$inserted = $wpdb->insert( self::table_name(), $row );
		if ( false === $inserted ) {
			// A concurrent replay may have won the unique-key race.
			$existing = self::get_by_idempotency_key( $idempotency_key );
			if ( $existing && hash_equals( (string) $existing['request_hash'], $request_hash ) ) {
				$existing['created_now'] = false;
				return $existing;
			}
			throw new RuntimeException( 'byok_job_create_failed' );
		}

		$created = self::get( $job_id );
		if ( ! $created ) {
			throw new RuntimeException( 'byok_job_create_failed' );
		}
		$created['created_now'] = true;
		return $created;
	}

	public static function get( string $job_id ): ?array {
		global $wpdb;
		$row = $wpdb->get_row(
			$wpdb->prepare( 'SELECT * FROM ' . self::table_name() . ' WHERE job_id=%s LIMIT 1', $job_id ),
			ARRAY_A
		);
		return is_array( $row ) ? $row : null;
	}

	public static function get_by_idempotency_key( string $key ): ?array {
		global $wpdb;
		$row = $wpdb->get_row(
			$wpdb->prepare( 'SELECT * FROM ' . self::table_name() . ' WHERE idempotency_key=%s LIMIT 1', $key ),
			ARRAY_A
		);
		return is_array( $row ) ? $row : null;
	}

	public static function current( int $production_id, int $agent_id = 0 ): ?array {
		global $wpdb;
		$states       = self::sql_placeholders( self::OPEN_STATES );
		$args         = array_merge( array( $production_id ), self::OPEN_STATES );
		$agent_clause = '';
		if ( $agent_id > 0 ) {
			$agent_clause = ' AND agent_id=%d';
			array_splice( $args, 1, 0, array( $agent_id ) );
		}
		$sql = 'SELECT * FROM ' . self::table_name() . " WHERE production_id=%d{$agent_clause} AND superseded_at IS NULL AND status IN ({$states}) ORDER BY id DESC LIMIT 1";
		$row = $wpdb->get_row( $wpdb->prepare( $sql, $args ), ARRAY_A );
		return is_array( $row ) ? $row : null;
	}

	public static function has_live_job( int $production_id ): bool {
		return null !== self::current( $production_id );
	}

	/** Latest authoritative attempt, including a terminal waiting to be applied. */
	public static function latest_attempt( int $production_id, int $agent_id ): ?array {
		global $wpdb;
		$row = $wpdb->get_row(
			$wpdb->prepare(
				'SELECT * FROM ' . self::table_name() . ' WHERE production_id=%d AND agent_id=%d AND superseded_at IS NULL ORDER BY id DESC LIMIT 1',
				$production_id,
				$agent_id
			),
			ARRAY_A
		);
		return is_array( $row ) ? $row : null;
	}

	/** Latest attempt including superseded history, used to rotate identity. */
	public static function latest_any_attempt( int $production_id, int $agent_id ): ?array {
		global $wpdb;
		$row = $wpdb->get_row(
			$wpdb->prepare(
				'SELECT * FROM ' . self::table_name() . ' WHERE production_id=%d AND agent_id=%d ORDER BY id DESC LIMIT 1',
				$production_id,
				$agent_id
			),
			ARRAY_A
		);
		return is_array( $row ) ? $row : null;
	}

	/** A durable attempt means the watchdog must nudge readback, never fail it. */
	public static function has_pending_application( int $production_id ): bool {
		global $wpdb;
		return (bool) $wpdb->get_var(
			$wpdb->prepare(
				'SELECT 1 FROM ' . self::table_name() . ' j LEFT JOIN ' . $wpdb->prefix . Costudio_Production::TABLE . " p ON p.id=j.production_id WHERE j.production_id=%d AND j.superseded_at IS NULL AND (j.status IN ('queued','processing','cancel_requested') OR (j.status IN ('completed','failed','cancelled','uncertain') AND j.applied_at IS NULL) OR (j.status='completed' AND p.current_agent=j.agent_id)) ORDER BY j.id DESC LIMIT 1",
				$production_id
			)
		);
	}

	public static function mark_applied( string $job_id ): bool {
		global $wpdb;
		$now = current_time( 'mysql', true );
		return 1 === (int) $wpdb->query(
			$wpdb->prepare(
				'UPDATE ' . self::table_name() . " SET applied_at=%s, updated_at=%s WHERE job_id=%s AND status IN ('completed','failed','cancelled','uncertain') AND applied_at IS NULL",
				$now,
				$now,
				$job_id
			)
		);
	}

	/**
	 * Retire attempts from an agent onward before a manual retry or internal
	 * Reviewer rewind. Running requests are cancelled but retained for audit.
	 */
	public static function supersede_from( int $production_id, int $from_agent, string $source = 'retry' ): int {
		global $wpdb;
		$rows = $wpdb->get_results(
			$wpdb->prepare(
				'SELECT job_id,status FROM ' . self::table_name() . ' WHERE production_id=%d AND agent_id >= %d AND superseded_at IS NULL ORDER BY id ASC',
				$production_id,
				max( 1, $from_agent )
			),
			ARRAY_A
		);
		$count = 0;
		$now   = current_time( 'mysql', true );
		foreach ( $rows ?: array() as $row ) {
			$job_id = (string) $row['job_id'];
			$status = (string) $row['status'];
			if ( 'queued' === $status ) {
				self::transition( $job_id, array( 'queued' ), 'superseded', array( 'terminal_source' => sanitize_key( $source ) ) );
			} elseif ( 'processing' === $status ) {
				self::request_cancel( $job_id, $source );
			}
			$updated = $wpdb->query(
				$wpdb->prepare(
					'UPDATE ' . self::table_name() . ' SET superseded_at=%s, updated_at=%s WHERE job_id=%s AND superseded_at IS NULL',
					$now,
					$now,
					$job_id
				)
			);
			$count += 1 === (int) $updated ? 1 : 0;
		}
		return $count;
	}

	public static function request_cancel_for_production( int $production_id, string $source = 'user' ): int {
		global $wpdb;
		$rows = $wpdb->get_col(
			$wpdb->prepare(
				'SELECT job_id FROM ' . self::table_name() . " WHERE production_id=%d AND superseded_at IS NULL AND status IN ('queued','processing')",
				$production_id
			)
		);
		$count = 0;
		foreach ( $rows ?: array() as $job_id ) {
			$count += self::request_cancel( (string) $job_id, $source ) ? 1 : 0;
		}
		return $count;
	}

	/**
	 * Apply a conditional state transition. Terminal states are immutable.
	 */
	public static function transition( string $job_id, array $expected, string $next, array $data = array(), array $conditions = array() ): bool {
		global $wpdb;

		$next     = sanitize_key( $next );
		$expected = array_values( array_unique( array_map( 'sanitize_key', $expected ) ) );
		if ( ! in_array( $next, self::ALLOWED_STATES, true ) || empty( $expected ) ) {
			return false;
		}
		foreach ( $expected as $state ) {
			if ( ! self::is_valid_transition( $state, $next ) ) {
				return false;
			}
		}

		$allowed_data = self::sanitize_transition_data( $data );
		$allowed_data['status']     = $next;
		$allowed_data['updated_at'] = current_time( 'mysql', true );
		if ( in_array( $next, self::TERMINAL_STATES, true ) ) {
			$allowed_data['finished_at'] = $allowed_data['finished_at'] ?? current_time( 'mysql', true );
		}
		if ( 'superseded' === $next ) {
			$allowed_data['superseded_at'] = $allowed_data['superseded_at'] ?? current_time( 'mysql', true );
		}

		$set_parts = array();
		$set_args  = array();
		foreach ( $allowed_data as $column => $value ) {
			if ( null === $value ) {
				$set_parts[] = "{$column}=NULL";
				continue;
			}
			$set_parts[] = "{$column}=%s";
			$set_args[]  = $value;
		}
		$state_placeholders = self::sql_placeholders( $expected );
		$args = array_merge( $set_args, array( $job_id ), $expected );
		$sql  = 'UPDATE ' . self::table_name() . ' SET ' . implode( ', ', $set_parts )
			. " WHERE job_id=%s AND status IN ({$state_placeholders})";

		$condition_columns = array( 'worker_id', 'lease_expires_at', 'provider_dispatched_at' );
		foreach ( $condition_columns as $column ) {
			if ( ! array_key_exists( $column, $conditions ) ) {
				continue;
			}
			if ( null === $conditions[ $column ] ) {
				$sql .= " AND {$column} IS NULL";
				continue;
			}
			$sql   .= " AND {$column}=%s";
			$args[] = (string) $conditions[ $column ];
		}

		return 1 === (int) $wpdb->query( $wpdb->prepare( $sql, $args ) );
	}

	public static function is_valid_transition( string $from, string $to ): bool {
		$map = array(
			'queued'           => array( 'processing', 'cancelled', 'superseded' ),
			'processing'       => array( 'queued', 'completed', 'failed', 'cancel_requested', 'cancelled', 'uncertain', 'superseded' ),
			'cancel_requested' => array( 'cancelled', 'uncertain', 'superseded' ),
		);
		return isset( $map[ $from ] ) && in_array( $to, $map[ $from ], true );
	}

	/**
	 * Atomically claim the oldest available queued job.
	 */
	public static function claim_next( string $worker_id, int $lease_seconds = 660, string $driver = self::MODE_EXTERNAL_CLI ): ?array {
		global $wpdb;
		$driver = self::normalize_mode( $driver );
		if ( self::MODE_LEGACY === $driver ) {
			return null;
		}
		if ( ! self::acquire_advisory_lock( self::CLAIM_LOCK_NAME, 2 ) ) {
			return null;
		}

		try {
			$now = current_time( 'mysql', true );
			$row = $wpdb->get_row(
				$wpdb->prepare(
					'SELECT * FROM ' . self::table_name() . " WHERE status='queued' AND driver=%s AND available_at <= %s ORDER BY available_at ASC, id ASC LIMIT 1",
					$driver,
					$now
				),
				ARRAY_A
			);
			if ( ! is_array( $row ) ) {
				return null;
			}

			$lease = gmdate( 'Y-m-d H:i:s', time() + max( 60, $lease_seconds ) );
			$ok = self::transition(
				(string) $row['job_id'],
				array( 'queued' ),
				'processing',
				array(
					'worker_id'             => sanitize_text_field( $worker_id ),
					'claim_count'           => (int) $row['claim_count'] + 1,
					'claimed_at'            => $now,
					'processing_started_at' => $row['processing_started_at'] ?: $now,
					'heartbeat_at'          => $now,
					'lease_expires_at'      => $lease,
				)
			);
			return $ok ? self::get( (string) $row['job_id'] ) : null;
		} finally {
			self::release_advisory_lock( self::CLAIM_LOCK_NAME );
		}
	}

	public static function available_queue_depth( string $driver = self::MODE_EXTERNAL_CLI ): int {
		global $wpdb;
		$driver = self::normalize_mode( $driver );
		if ( self::MODE_LEGACY === $driver ) {
			return 0;
		}
		return (int) $wpdb->get_var(
			$wpdb->prepare(
				'SELECT COUNT(*) FROM ' . self::table_name() . " WHERE status='queued' AND driver=%s AND available_at <= %s",
				$driver,
				current_time( 'mysql', true )
			)
		);
	}

	public static function heartbeat_job( string $job_id, string $worker_id, int $lease_seconds = 660 ): bool {
		global $wpdb;
		$now   = current_time( 'mysql', true );
		$lease = gmdate( 'Y-m-d H:i:s', time() + max( 60, $lease_seconds ) );
		return 1 === (int) $wpdb->query(
			$wpdb->prepare(
				'UPDATE ' . self::table_name() . " SET heartbeat_at=%s, lease_expires_at=%s, updated_at=%s WHERE job_id=%s AND worker_id=%s AND status IN ('processing','cancel_requested')",
				$now,
				$lease,
				$now,
				$job_id,
				$worker_id
			)
		);
	}

	public static function mark_provider_dispatched( string $job_id, string $worker_id ): bool {
		global $wpdb;
		$now = current_time( 'mysql', true );
		return 1 === (int) $wpdb->query(
			$wpdb->prepare(
				'UPDATE ' . self::table_name() . " SET provider_dispatched_at=%s, updated_at=%s WHERE job_id=%s AND worker_id=%s AND status='processing' AND provider_dispatched_at IS NULL",
				$now,
				$now,
				$job_id,
				$worker_id
			)
		);
	}

	public static function mark_response_received( string $job_id, string $worker_id ): bool {
		global $wpdb;
		$now = current_time( 'mysql', true );
		return 1 === (int) $wpdb->query(
			$wpdb->prepare(
				'UPDATE ' . self::table_name() . " SET first_byte_at=COALESCE(first_byte_at,%s), response_received_at=%s, updated_at=%s WHERE job_id=%s AND worker_id=%s AND status IN ('processing','cancel_requested') AND response_received_at IS NULL",
				$now,
				$now,
				$now,
				$job_id,
				$worker_id
			)
		);
	}

	public static function request_cancel( string $job_id, string $source = 'operator' ): bool {
		$job = self::get( $job_id );
		if ( ! $job ) {
			return false;
		}
		if ( 'queued' === (string) $job['status'] ) {
			return self::transition(
				$job_id,
				array( 'queued' ),
				'cancelled',
				array( 'terminal_source' => sanitize_key( $source ), 'retry_safe' => 1 )
			);
		}
		if ( 'processing' === (string) $job['status'] ) {
			return self::transition(
				$job_id,
				array( 'processing' ),
				'cancel_requested',
				array( 'terminal_source' => sanitize_key( $source ) )
			);
		}
		return false;
	}

	public static function mark_usage_recorded( string $job_id ): bool {
		global $wpdb;
		$now = current_time( 'mysql', true );
		return 1 === (int) $wpdb->query(
			$wpdb->prepare(
				'UPDATE ' . self::table_name() . ' SET usage_recorded_at=%s, updated_at=%s WHERE job_id=%s AND usage_recorded_at IS NULL',
				$now,
				$now,
				$job_id
			)
		);
	}

	public static function usage_reconciliation_rows( int $limit = 20 ): array {
		global $wpdb;
		$limit = max( 1, min( 100, $limit ) );
		return (array) $wpdb->get_results(
			$wpdb->prepare(
				'SELECT job_id,production_id,agent_id,provider,model,processing_started_at,finished_at,usage_json,usage_recorded_at FROM ' . self::table_name() . " WHERE status IN ('completed','cancelled','uncertain') AND response_received_at IS NOT NULL AND usage_json IS NOT NULL AND usage_recorded_at IS NULL ORDER BY id ASC LIMIT %d",
				$limit
			),
			ARRAY_A
		);
	}

	public static function acquire_worker_lock( int $wait_seconds = 0 ): bool {
		return self::acquire_advisory_lock( self::WORKER_LOCK_NAME, $wait_seconds );
	}

	public static function release_worker_lock(): void {
		self::release_advisory_lock( self::WORKER_LOCK_NAME );
	}

	/**
	 * Reconcile expired processing leases without ever replaying an ambiguous
	 * provider request.
	 */
	public static function recover_expired_leases( int $limit = 20 ): array {
		global $wpdb;
		$limit = max( 1, min( 100, $limit ) );
		$rows  = $wpdb->get_results(
			$wpdb->prepare(
				'SELECT * FROM ' . self::table_name() . " WHERE status IN ('processing','cancel_requested') AND lease_expires_at IS NOT NULL AND lease_expires_at < %s ORDER BY lease_expires_at ASC LIMIT %d",
				current_time( 'mysql', true ),
				$limit
			),
			ARRAY_A
		);
		$result = array( 'requeued' => 0, 'completed' => 0, 'uncertain' => 0, 'cancelled' => 0 );

		foreach ( $rows ?: array() as $row ) {
			if ( self::recover_persisted_pipeline_result( $row ) ) {
				$result['completed']++;
				continue;
			}
			$action = self::lease_recovery_action( $row );
			$conditions = array(
				'worker_id'             => (string) $row['worker_id'],
				'lease_expires_at'      => (string) $row['lease_expires_at'],
				'provider_dispatched_at' => empty( $row['provider_dispatched_at'] ) ? null : (string) $row['provider_dispatched_at'],
			);
			if ( 'requeue' === $action ) {
				$ok = self::transition(
					(string) $row['job_id'],
					array( 'processing' ),
					'queued',
					array(
						'worker_id'             => '',
						'available_at'          => current_time( 'mysql', true ),
						'claimed_at'            => null,
						'processing_started_at' => null,
						'heartbeat_at'          => null,
						'lease_expires_at'      => null,
						'terminal_source'       => 'lease_recovery',
					),
					$conditions
				);
				$result['requeued'] += $ok ? 1 : 0;
			} elseif ( 'cancel' === $action ) {
				$ok = self::transition(
					(string) $row['job_id'],
					array( 'cancel_requested' ),
					'cancelled',
					array( 'terminal_source' => 'lease_recovery', 'retry_safe' => 1 ),
					$conditions
				);
				$result['cancelled'] += $ok ? 1 : 0;
			} else {
				$ok = self::transition(
					(string) $row['job_id'],
					array( 'processing', 'cancel_requested' ),
					'uncertain',
					array(
						'error_code'     => 'provider_result_uncertain',
						'error_public'   => __( 'The local BYOK worker stopped after the provider request was sent. The request was not repeated automatically because provider usage may have occurred.', 'contentosapp-studio' ),
						'retry_safe'     => 0,
						'terminal_source'=> 'lease_recovery',
					),
					$conditions
				);
				$result['uncertain'] += $ok ? 1 : 0;
			}
		}

		return $result;
	}

	/**
	 * A pipeline-agent job may have durably written its production output just
	 * before the CLI process died. That output is authoritative: finalize the
	 * queue row instead of reporting an ambiguous provider result or replaying it.
	 */
	private static function recover_persisted_pipeline_result( array $job ): bool {
		try {
			$payload = Costudio_BYOK_Job_Codec::decode_request( (string) ( $job['request_ciphertext'] ?? '' ) );
			if ( 'pipeline_agent' !== (string) ( $payload['execution_kind'] ?? '' ) ) {
				return false;
			}
			$production_id = (int) $job['production_id'];
			$agent_id      = (int) $job['agent_id'];
			$production    = Costudio_Production::get_by_id( $production_id );
			$column        = $production ? $production->map_step_to_column( $agent_id ) : '';
			$text          = $production && $column ? trim( (string) ( $production->{$column} ?? '' ) ) : '';
			if ( '' === $text ) {
				return false;
			}

			$usage = Costudio_Usage::get_agent_usage( $production_id, $agent_id ) ?: array();
			$result_ciphertext = Costudio_BYOK_Job_Codec::encode_result( array(
				'text'     => $text,
				'provider' => (string) $job['provider'],
				'model'    => (string) ( $usage['model'] ?? $job['model'] ),
				'usage'    => Costudio_BYOK_Job_Codec::normalize_usage( $usage ),
			) );
			$usage_json = wp_json_encode( array_merge(
				Costudio_BYOK_Job_Codec::normalize_usage( $usage ),
				array( 'latency_ms' => max( 0, (int) ( $usage['latency_ms'] ?? 0 ) ) )
			) );
			$won = self::transition(
				(string) $job['job_id'],
				array( 'processing' ),
				'completed',
				array(
					'result_ciphertext' => $result_ciphertext,
					'usage_json'        => $usage_json,
					'retry_safe'        => 1,
					'terminal_source'   => 'lease_recovery_persisted_output',
				),
				array(
					'worker_id'        => (string) $job['worker_id'],
					'lease_expires_at' => (string) $job['lease_expires_at'],
				)
			);
			if ( $won ) {
				if ( Costudio_Usage::has_byok_job( (string) $job['job_id'] ) ) {
					self::mark_usage_recorded( (string) $job['job_id'] );
				}
				if ( class_exists( 'Costudio_Runner' ) ) {
					Costudio_Runner::resume_durable_result( $production_id, 'lease_recovery' );
				}
			}
			return $won;
		} catch ( Throwable $error ) {
			return false;
		}
	}

	public static function lease_recovery_action( array $job ): string {
		if ( 'cancel_requested' === (string) ( $job['status'] ?? '' ) && empty( $job['provider_dispatched_at'] ) ) {
			return 'cancel';
		}
		return empty( $job['provider_dispatched_at'] ) ? 'requeue' : 'uncertain';
	}

	public static function record_worker_heartbeat( array $data ): void {
		$safe = array(
			'worker_id'  => sanitize_text_field( (string) ( $data['worker_id'] ?? '' ) ),
			'pid'        => max( 0, (int) ( $data['pid'] ?? 0 ) ),
			'sapi'       => sanitize_key( (string) ( $data['sapi'] ?? PHP_SAPI ) ),
			'state'      => sanitize_key( (string) ( $data['state'] ?? 'idle' ) ),
			'job_id'     => sanitize_text_field( (string) ( $data['job_id'] ?? '' ) ),
			'timestamp'  => time(),
			'updated_at' => current_time( 'mysql', true ),
		);
		update_option( self::WORKER_OPTION, $safe, false );
	}

	public static function status_summary(): array {
		global $wpdb;
		$counts = array();
		$rows = $wpdb->get_results( 'SELECT status, COUNT(*) AS total FROM ' . self::table_name() . ' GROUP BY status', ARRAY_A );
		foreach ( $rows ?: array() as $row ) {
			$counts[ (string) $row['status'] ] = (int) $row['total'];
		}

		return array(
			'mode'               => self::mode(),
			'queue_version'      => self::QUEUE_VERSION,
			'pipeline_connected' => self::pipeline_connected(),
			'canary_eligible'    => self::current_license_is_canary(),
			'new_jobs_durable'   => self::routes_new_jobs(),
			'canary_count'       => count( self::canary_hashes() ),
			'queue_depth'        => (int) ( $counts['queued'] ?? 0 ),
			'counts'             => $counts,
			'worker'             => get_option( self::WORKER_OPTION, array() ),
			'studio_version'     => defined( 'COSTUDIO_VERSION' ) ? COSTUDIO_VERSION : '',
			'db_version'         => class_exists( 'Costudio_DB' ) ? Costudio_DB::DB_VERSION : '',
		);
	}

	public static function list_rows( int $limit = 20, string $status = '' ): array {
		global $wpdb;
		$limit = max( 1, min( 200, $limit ) );
		$columns = 'job_id,status,production_id,production_uid,agent_id,attempt_id,provider,model,driver,claim_count,worker_id,available_at,claimed_at,processing_started_at,heartbeat_at,lease_expires_at,provider_dispatched_at,first_byte_at,response_received_at,error_code,retry_safe,terminal_source,applied_at,usage_recorded_at,created_at,updated_at,finished_at,payload_purged_at';
		if ( in_array( $status, self::ALLOWED_STATES, true ) ) {
			return (array) $wpdb->get_results(
				$wpdb->prepare( 'SELECT ' . $columns . ' FROM ' . self::table_name() . ' WHERE status=%s ORDER BY id DESC LIMIT %d', $status, $limit ),
				ARRAY_A
			);
		}
		return (array) $wpdb->get_results(
			$wpdb->prepare( 'SELECT ' . $columns . ' FROM ' . self::table_name() . ' ORDER BY id DESC LIMIT %d', $limit ),
			ARRAY_A
		);
	}

	/**
	 * Purge sensitive payloads after seven days and audit rows after 90 days.
	 */
	public static function cleanup(): array {
		global $wpdb;
		$table      = self::table_name();
		$purge_cut  = gmdate( 'Y-m-d H:i:s', time() - ( 7 * DAY_IN_SECONDS ) );
		$delete_cut = gmdate( 'Y-m-d H:i:s', time() - ( 90 * DAY_IN_SECONDS ) );
		$terminals  = self::sql_placeholders( self::TERMINAL_STATES );

		$purge_args = array_merge( array( current_time( 'mysql', true ), $purge_cut ), self::TERMINAL_STATES );
		$purged = (int) $wpdb->query(
			$wpdb->prepare(
				"UPDATE {$table} SET request_ciphertext=NULL, partial_ciphertext=NULL, result_ciphertext=NULL, usage_json=NULL, payload_purged_at=%s WHERE finished_at < %s AND status IN ({$terminals}) AND payload_purged_at IS NULL",
				$purge_args
			)
		);

		$delete_args = array_merge( array( $delete_cut ), self::TERMINAL_STATES );
		$deleted = (int) $wpdb->query(
			$wpdb->prepare(
				"DELETE FROM {$table} WHERE finished_at < %s AND status IN ({$terminals})",
				$delete_args
			)
		);

		return array( 'payloads_purged' => max( 0, $purged ), 'rows_deleted' => max( 0, $deleted ) );
	}

	private static function table_name(): string {
		global $wpdb;
		return $wpdb->prefix . self::TABLE;
	}

	private static function acquire_advisory_lock( string $name, int $wait_seconds ): bool {
		global $wpdb;
		$key = 'costudio:' . substr( hash( 'sha256', $name ), 0, 48 );
		return 1 === (int) $wpdb->get_var( $wpdb->prepare( 'SELECT GET_LOCK(%s, %d)', $key, max( 0, $wait_seconds ) ) );
	}

	private static function release_advisory_lock( string $name ): void {
		global $wpdb;
		$key = 'costudio:' . substr( hash( 'sha256', $name ), 0, 48 );
		$wpdb->get_var( $wpdb->prepare( 'SELECT RELEASE_LOCK(%s)', $key ) );
	}

	private static function sanitize_transition_data( array $data ): array {
		$allowed = array(
			'available_at', 'worker_id', 'claim_count', 'claimed_at', 'processing_started_at',
			'heartbeat_at', 'lease_expires_at', 'provider_dispatched_at', 'first_byte_at',
			'response_received_at', 'partial_ciphertext', 'result_ciphertext', 'usage_json',
			'error_code', 'error_public', 'retry_safe', 'terminal_source', 'applied_at', 'usage_recorded_at',
			'superseded_at', 'finished_at',
		);
		$result = array();
		foreach ( $allowed as $column ) {
			if ( array_key_exists( $column, $data ) ) {
				$result[ $column ] = $data[ $column ];
			}
		}
		return $result;
	}

	private static function sql_placeholders( array $items ): string {
		return implode( ',', array_fill( 0, count( $items ), '%s' ) );
	}

	private static function normalize_datetime( string $value, string $fallback ): string {
		if ( '' === $value ) {
			return $fallback;
		}
		$timestamp = strtotime( $value . ' UTC' );
		return false === $timestamp ? $fallback : gmdate( 'Y-m-d H:i:s', $timestamp );
	}

	private static function hash_field( string $value, string $name ): string {
		$value = strtolower( trim( $value ) );
		if ( 1 !== preg_match( '/^[a-f0-9]{64}$/', $value ) ) {
			throw new RuntimeException( 'byok_job_invalid_' . $name );
		}
		return $value;
	}

	private static function optional_hash( string $value ): string {
		$value = strtolower( trim( $value ) );
		return 1 === preg_match( '/^[a-f0-9]{64}$/', $value ) ? $value : '';
	}

	private static function uuid_field( string $value, string $name ): string {
		$value = strtolower( trim( $value ) );
		if ( 1 !== preg_match( '/^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$/', $value ) ) {
			throw new RuntimeException( 'byok_job_invalid_' . $name );
		}
		return $value;
	}
}
