<?php
/** Durable singleton worker for local BYOK provider requests. */

defined( 'ABSPATH' ) || exit;

class Costudio_BYOK_Worker {

	public const DEFAULT_MAX_RUNTIME  = 720;
	public const DEFAULT_IDLE_TIMEOUT = 90;
	public const DEFAULT_POLL_INTERVAL = 2;
	public const DEFAULT_LEASE_SECONDS = 660;
	private const MIN_JOB_BUDGET       = 570;

	/** @var Costudio_BYOK_Provider_Executor */
	private $executor;

	public function __construct( ?Costudio_BYOK_Provider_Executor $executor = null ) {
		$this->executor = $executor ?: new Costudio_BYOK_Provider_Executor();
	}

	/**
	 * Run one singleton worker process.
	 *
	 * @return array{started:bool,processed:int,elapsed:int,reason:string,worker_id:string}
	 */
	public function run( array $options = array() ): array {
		$max_runtime  = max( 5, min( 900, (int) ( $options['max_runtime'] ?? self::DEFAULT_MAX_RUNTIME ) ) );
		$idle_timeout = max( 1, min( $max_runtime, (int) ( $options['idle_timeout'] ?? self::DEFAULT_IDLE_TIMEOUT ) ) );
		$poll_interval = max( 1, min( 10, (int) ( $options['poll_interval'] ?? self::DEFAULT_POLL_INTERVAL ) ) );
		$lease_seconds = max( 600, min( 900, (int) ( $options['lease_seconds'] ?? self::DEFAULT_LEASE_SECONDS ) ) );
		$worker_id    = self::new_worker_id();
		$started_at   = microtime( true );
		$last_work_at = $started_at;
		$processed    = 0;
		$reason       = 'max_runtime';

		if ( ! Costudio_BYOK_Jobs::acquire_worker_lock( 0 ) ) {
			self::log( 'info', 'byok_jobs.worker_lock_busy', array( 'worker_id' => $worker_id ) );
			return array( 'started' => false, 'processed' => 0, 'elapsed' => 0, 'reason' => 'lock_busy', 'worker_id' => $worker_id );
		}

		self::log( 'info', 'byok_jobs.worker_started', array(
			'worker_id' => $worker_id,
			'pid' => function_exists( 'getmypid' ) ? (int) getmypid() : 0,
			'php_sapi' => PHP_SAPI,
			'max_runtime' => $max_runtime,
		) );

		try {
			Costudio_BYOK_Jobs::cleanup();
			while ( true ) {
				$elapsed   = (int) floor( microtime( true ) - $started_at );
				$remaining = $max_runtime - $elapsed;
				if ( $remaining <= 0 ) {
					$reason = 'max_runtime';
					break;
				}

				$this->recover_leases();
				$this->reconcile_usage();

				Costudio_BYOK_Jobs::record_worker_heartbeat( self::heartbeat_payload( $worker_id, 'idle', '' ) );
				if ( $remaining < self::MIN_JOB_BUDGET ) {
					if ( Costudio_BYOK_Jobs::available_queue_depth( Costudio_BYOK_Jobs::MODE_EXTERNAL_CLI ) > 0 ) {
						$reason = 'insufficient_job_budget';
						break;
					}
					if ( ( microtime( true ) - $last_work_at ) >= $idle_timeout ) {
						$reason = 'idle_timeout';
						break;
					}
					usleep( $poll_interval * 1000000 );
					continue;
				}

				// Also drain durable final-publication jobs when this optional CLI
				// worker exists. Regular sites retain the automatic web fallback.
				if ( class_exists( 'Costudio_Publication_Jobs' ) && Costudio_Publication_Jobs::queue_depth() > 0 ) {
					if ( Costudio_Publication_Worker::run_next_cli( $worker_id ) ) {
						$last_work_at = microtime( true );
						$processed++;
						continue;
					}
				}

				$job = Costudio_BYOK_Jobs::claim_next( $worker_id, $lease_seconds, Costudio_BYOK_Jobs::MODE_EXTERNAL_CLI );
				if ( ! $job ) {
					if ( ( microtime( true ) - $last_work_at ) >= $idle_timeout ) {
						$reason = 'idle_timeout';
						break;
					}
					usleep( $poll_interval * 1000000 );
					continue;
				}

				$last_work_at = microtime( true );
				$processed++;
				$this->process_job( $job, $worker_id, $lease_seconds );
				$last_work_at = microtime( true );
			}
		} finally {
			$this->reconcile_usage();
			Costudio_BYOK_Jobs::record_worker_heartbeat( self::heartbeat_payload( $worker_id, 'stopped', '' ) );
			Costudio_BYOK_Jobs::release_worker_lock();
			self::log( 'info', 'byok_jobs.worker_stopped', array(
				'worker_id' => $worker_id,
				'processed' => $processed,
				'elapsed' => max( 0, (int) floor( microtime( true ) - $started_at ) ),
				'reason' => $reason,
			) );
		}

		return array(
			'started'   => true,
			'processed' => $processed,
			'elapsed'   => max( 0, (int) floor( microtime( true ) - $started_at ) ),
			'reason'    => $reason,
			'worker_id' => $worker_id,
		);
	}

	private function process_job( array $job, string $worker_id, int $lease_seconds ): void {
		$job_id       = (string) $job['job_id'];
		$production_id = (int) $job['production_id'];
		$agent_id     = (int) $job['agent_id'];
		$provider     = (string) $job['provider'];
		$provider_dispatched = false;
		$response_received   = false;
		$provider_completed  = false;
		$result               = null;
		$execution_kind       = 'provider_request';

		Costudio_BYOK_Jobs::record_worker_heartbeat( self::heartbeat_payload( $worker_id, 'processing', $job_id ) );
		self::log( 'info', 'byok_jobs.claimed', self::job_context( $job, $worker_id ), $production_id, $agent_id );

		try {
			$payload = Costudio_BYOK_Job_Codec::decode_request( (string) $job['request_ciphertext'] );
			$execution_kind = (string) ( $payload['execution_kind'] ?? 'provider_request' );
			$before_dispatch = function () use ( $job_id, $worker_id, $production_id, $agent_id, $provider, &$provider_dispatched ): void {
					$fresh = Costudio_BYOK_Jobs::get( $job_id );
					if ( $fresh && 'cancel_requested' === (string) $fresh['status'] ) {
						Costudio_BYOK_Jobs::transition( $job_id, array( 'cancel_requested' ), 'cancelled', array(
							'terminal_source' => 'worker_pre_dispatch_cancel',
							'retry_safe' => 1,
						) );
						throw new Costudio_BYOK_Provider_Executor_Exception( 'byok_cancelled_before_dispatch' );
					}
					if ( ! Costudio_BYOK_Jobs::mark_provider_dispatched( $job_id, $worker_id ) ) {
						throw new Costudio_BYOK_Provider_Executor_Exception( 'byok_dispatch_checkpoint_refused' );
					}
					$provider_dispatched = true;
					self::log( 'info', 'byok_jobs.provider_dispatched', array(
						'job_id' => $job_id, 'worker_id' => $worker_id, 'provider' => $provider,
					), $production_id, $agent_id );
				};
			$response_checkpoint = function () use ( $job_id, $worker_id, &$response_received ): void {
					if ( ! Costudio_BYOK_Jobs::mark_response_received( $job_id, $worker_id ) ) {
						throw new Costudio_BYOK_Provider_Executor_Exception( 'byok_response_checkpoint_refused' );
					}
					$response_received = true;
				};
			$heartbeat = function ( string $stage ) use ( $job_id, $worker_id, $lease_seconds ): void {
					Costudio_BYOK_Jobs::heartbeat_job( $job_id, $worker_id, $lease_seconds );
					Costudio_BYOK_Jobs::record_worker_heartbeat( self::heartbeat_payload( $worker_id, sanitize_key( $stage ), $job_id ) );
				};

			if ( 'pipeline_agent' === $execution_kind ) {
				$heartbeat( 'pipeline_preparing' );
				$pipeline = new Costudio_Pipeline( $payload );
				$pipeline->configure_durable_worker(
					$job_id,
					$before_dispatch,
					$response_checkpoint,
					(int) ( $payload['deadline_seconds'] ?? 540 )
				);
				$pipeline->run_step( $agent_id, $production_id );
				$provider_completed = $response_received;
				$heartbeat( 'pipeline_persisted' );

				$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 ) {
					throw new Costudio_BYOK_Provider_Executor_Exception( 'byok_pipeline_output_missing' );
				}
				$usage_row = Costudio_Usage::get_agent_usage( $production_id, $agent_id ) ?: array();
				$result = array(
					'text'       => $text,
					'provider'   => $provider,
					'model'      => (string) ( $usage_row['model'] ?? $job['model'] ),
					'usage'      => Costudio_BYOK_Job_Codec::normalize_usage( $usage_row ),
					'elapsed_ms' => max( 0, (int) ( $usage_row['latency_ms'] ?? 0 ) ),
				);
			} else {
				$result = $this->executor->execute(
					$payload,
					$before_dispatch,
					$response_checkpoint,
					$heartbeat
				);
			}
			if ( 'provider_request' === $execution_kind ) {
				$provider_completed = true;
			}

			$usage = Costudio_BYOK_Job_Codec::normalize_usage( (array) $result['usage'] );
			$usage['latency_ms'] = max( 0, (int) $result['elapsed_ms'] );
			$usage['grounding'] = ! empty( $payload['has_grounding'] );
			$usage_json = wp_json_encode( $usage );
			$fresh = Costudio_BYOK_Jobs::get( $job_id );
			if ( $fresh && 'cancel_requested' === (string) $fresh['status'] ) {
				$won = Costudio_BYOK_Jobs::transition( $job_id, array( 'cancel_requested' ), 'cancelled', array(
					'usage_json' => $usage_json,
					'terminal_source' => 'worker_post_dispatch_cancel',
					'retry_safe' => 0,
				) );
				if ( $won ) {
					$this->reconcile_usage_for_job( $job_id );
				}
				return;
			}

			$ciphertext = Costudio_BYOK_Job_Codec::encode_result( $result );
			$won = Costudio_BYOK_Jobs::transition( $job_id, array( 'processing' ), 'completed', array(
				'result_ciphertext' => $ciphertext,
				'usage_json' => $usage_json,
				'retry_safe' => 1,
				'terminal_source' => 'worker',
			) );
			if ( $won ) {
				self::log( 'info', 'byok_jobs.completed', self::job_context( $job, $worker_id ), $production_id, $agent_id );
				$this->reconcile_usage_for_job( $job_id );
				if ( class_exists( 'Costudio_Runner' ) ) {
					Costudio_Runner::resume_durable_result( $production_id, 'worker_completed' );
				}
			} else {
				$fresh = Costudio_BYOK_Jobs::get( $job_id );
				if ( $fresh && 'cancel_requested' === (string) $fresh['status'] ) {
					$cancelled = Costudio_BYOK_Jobs::transition( $job_id, array( 'cancel_requested' ), 'cancelled', array(
						'usage_json' => $usage_json,
						'retry_safe' => 0,
						'terminal_source' => 'worker_post_dispatch_cancel',
					) );
					if ( $cancelled ) {
						$this->reconcile_usage_for_job( $job_id );
					}
				} else {
					self::log( 'warn', 'byok_jobs.terminal_transition_refused', self::job_context( $job, $worker_id ), $production_id, $agent_id );
				}
			}
		} catch ( Throwable $error ) {
			$fresh = Costudio_BYOK_Jobs::get( $job_id );
			if ( ! $fresh || in_array( (string) $fresh['status'], array( 'completed', 'failed', 'cancelled', 'uncertain', 'superseded' ), true ) ) {
				return;
			}
			$provider_dispatched = $provider_dispatched || ! empty( $fresh['provider_dispatched_at'] );
			$response_received   = $response_received || ! empty( $fresh['response_received_at'] );
			if ( 'pipeline_agent' === $execution_kind && $response_received ) {
				$provider_completed = true;
				if ( ! is_array( $result ) ) {
					$usage_row = Costudio_Usage::get_agent_usage( $production_id, $agent_id ) ?: array();
					$result = array(
						'usage'      => Costudio_BYOK_Job_Codec::normalize_usage( $usage_row ),
						'elapsed_ms' => max( 0, (int) ( $usage_row['latency_ms'] ?? 0 ) ),
					);
				}
			}

			if ( 'cancel_requested' === (string) $fresh['status'] ) {
				$next = $provider_dispatched && ! $response_received ? 'uncertain' : 'cancelled';
				Costudio_BYOK_Jobs::transition( $job_id, array( 'cancel_requested' ), $next, array(
					'error_code' => 'cancelled_by_request',
					'error_public' => __( 'The BYOK request was cancelled.', 'contentosapp-studio' ),
					'retry_safe' => $provider_dispatched ? 0 : 1,
					'terminal_source' => 'worker_cancel',
				) );
				return;
			}

			$resolution = self::failure_resolution( $error, $provider_dispatched, $response_received, $provider_completed );
			$next       = $resolution['status'];
			$code       = $resolution['code'];
			$public     = $resolution['public'];
			$retry_safe = $resolution['retry_safe'];
			$failure_data = array(
				'error_code' => $code,
				'error_public' => $public,
				'retry_safe' => $retry_safe,
				'terminal_source' => 'worker',
			);
			if ( $provider_completed && is_array( $result ) ) {
				$usage = Costudio_BYOK_Job_Codec::normalize_usage( (array) ( $result['usage'] ?? array() ) );
				$usage['latency_ms'] = max( 0, (int) ( $result['elapsed_ms'] ?? 0 ) );
				$failure_data['usage_json'] = wp_json_encode( $usage );
			}

			$won = Costudio_BYOK_Jobs::transition( $job_id, array( 'processing' ), $next, $failure_data );
			if ( $won ) {
				self::log( 'uncertain' === $next ? 'warn' : 'error', 'byok_jobs.' . $next, array_merge(
					self::job_context( $job, $worker_id ),
					array( 'error_code' => $code, 'provider_dispatched' => $provider_dispatched, 'response_received' => $response_received )
				), $production_id, $agent_id );
				if ( class_exists( 'Costudio_Runner' ) ) {
					Costudio_Runner::resume_durable_result( $production_id, 'worker_' . $next );
				}
			}
		} finally {
			Costudio_BYOK_Jobs::record_worker_heartbeat( self::heartbeat_payload( $worker_id, 'idle', '' ) );
		}
	}

	/**
	 * Pure failure classifier used by the worker and deterministic harnesses.
	 * A successful provider result that could not be persisted is ambiguous for
	 * usage purposes and must never be replayed automatically.
	 *
	 * @return array{status:string,code:string,public:string,retry_safe:int}
	 */
	public static function failure_resolution( Throwable $error, bool $dispatched, bool $response_received, bool $provider_completed = false ): array {
		if ( $provider_completed ) {
			return array(
				'status' => 'uncertain',
				'code' => 'provider_result_persistence_uncertain',
				'public' => __( 'The BYOK provider returned a result, but the local worker could not persist it safely. The request was not repeated automatically.', 'contentosapp-studio' ),
				'retry_safe' => 0,
			);
		}
		if ( $dispatched && ! $response_received ) {
			return array(
				'status' => 'uncertain',
				'code' => 'provider_result_uncertain',
				'public' => __( 'The local BYOK worker stopped after sending the request to the AI provider. It was not repeated automatically because provider usage may have occurred.', 'contentosapp-studio' ),
				'retry_safe' => 0,
			);
		}
		$classification = Costudio_BYOK_Provider_Executor::classify_error( $error );
		return array(
			'status' => 'failed',
			'code' => (string) $classification['code'],
			'public' => self::public_error_message( (string) $classification['code'] ),
			'retry_safe' => ! empty( $classification['retry_safe'] ) ? 1 : 0,
		);
	}

	private function recover_leases(): void {
		$result = Costudio_BYOK_Jobs::recover_expired_leases( 20 );
		foreach ( array( 'requeued', 'uncertain', 'cancelled' ) as $state ) {
			if ( ! empty( $result[ $state ] ) ) {
				self::log( 'uncertain' === $state ? 'warn' : 'info', 'byok_jobs.lease_' . $state, array( 'count' => (int) $result[ $state ] ) );
			}
		}
	}

	private function reconcile_usage(): void {
		foreach ( Costudio_BYOK_Jobs::usage_reconciliation_rows( 20 ) as $row ) {
			$this->record_usage_row( $row );
		}
	}

	private function reconcile_usage_for_job( string $job_id ): void {
		$job = Costudio_BYOK_Jobs::get( $job_id );
		if ( $job ) {
			$this->record_usage_row( $job );
		}
	}

	private function record_usage_row( array $row ): void {
		$job_id = (string) ( $row['job_id'] ?? '' );
		if ( '' === $job_id || ! empty( $row['usage_recorded_at'] ) || empty( $row['usage_json'] ) ) {
			return;
		}
		if ( Costudio_Usage::has_byok_job( $job_id ) ) {
			Costudio_BYOK_Jobs::mark_usage_recorded( $job_id );
			return;
		}
		$usage = json_decode( (string) $row['usage_json'], true );
		if ( ! is_array( $usage ) ) {
			return;
		}
		$recorded = Costudio_Usage::record( array(
			'production_id' => (int) ( $row['production_id'] ?? 0 ),
			'agent' => (int) ( $row['agent_id'] ?? 0 ),
			'provider' => (string) ( $row['provider'] ?? '' ),
			'model' => (string) ( $row['model'] ?? '' ),
			'input_tokens' => (int) ( $usage['input_tokens'] ?? 0 ),
			'output_tokens' => (int) ( $usage['output_tokens'] ?? 0 ),
			'latency_ms' => (int) ( $usage['latency_ms'] ?? 0 ),
			'started_at' => (string) ( $row['processing_started_at'] ?? '' ),
			'grounding' => ! empty( $usage['grounding'] ),
			'byok_job_id' => $job_id,
		) );
		if ( $recorded || Costudio_Usage::has_byok_job( $job_id ) ) {
			Costudio_BYOK_Jobs::mark_usage_recorded( $job_id );
			self::log( 'info', 'byok_jobs.usage_reconciled', array( 'job_id' => $job_id ) );
		}
	}

	private static function public_error_message( string $code ): string {
		if ( 'provider_rate_limited' === $code ) {
			return __( 'The selected BYOK provider is temporarily rate-limiting requests. Retry this agent later.', 'contentosapp-studio' );
		}
		if ( 'provider_http_4xx' === $code ) {
			return __( 'The BYOK provider rejected this request. Check the API key, model and account limits before retrying.', 'contentosapp-studio' );
		}
		if ( 'provider_transport_timeout' === $code ) {
			return __( 'The BYOK provider connection ended without an authoritative response. The request was not repeated automatically.', 'contentosapp-studio' );
		}
		return __( 'The BYOK provider could not complete this agent. Review the provider configuration and retry when ready.', 'contentosapp-studio' );
	}

	private static function new_worker_id(): string {
		$host = function_exists( 'gethostname' ) ? (string) gethostname() : 'host';
		$pid  = function_exists( 'getmypid' ) ? (int) getmypid() : 0;
		return sanitize_text_field( $host . ':' . $pid . ':' . wp_generate_password( 12, false, false ) );
	}

	private static function heartbeat_payload( string $worker_id, string $state, string $job_id ): array {
		return array(
			'worker_id' => $worker_id,
			'pid' => function_exists( 'getmypid' ) ? (int) getmypid() : 0,
			'sapi' => PHP_SAPI,
			'state' => $state,
			'job_id' => $job_id,
		);
	}

	private static function job_context( array $job, string $worker_id ): array {
		return array(
			'job_id' => (string) ( $job['job_id'] ?? '' ),
			'worker_id' => $worker_id,
			'production_id' => (int) ( $job['production_id'] ?? 0 ),
			'agent_id' => (int) ( $job['agent_id'] ?? 0 ),
			'provider' => (string) ( $job['provider'] ?? '' ),
			'model' => (string) ( $job['model'] ?? '' ),
			'claim_count' => (int) ( $job['claim_count'] ?? 0 ),
		);
	}

	private static function log( string $level, string $event, array $context, ?int $production_id = null, ?int $agent_id = null ): void {
		if ( 'error' === $level ) {
			Costudio_Logger::error( $event, $context, $production_id, $agent_id );
		} elseif ( 'warn' === $level ) {
			Costudio_Logger::warn( $event, $context, $production_id, $agent_id );
		} else {
			Costudio_Logger::info( $event, $context, $production_id, $agent_id );
		}
	}
}
