<?php

namespace Fotogenic_AI;

if ( ! defined( 'ABSPATH' ) ) {
    exit;
}

final class Bulk_Run_Runner {
    private static ?Bulk_Run_Runner $instance = null;
    private Bulk_Repo $repo;
    private Universal_Bulk_Item_Dispatcher $universal_item_dispatcher;
    private const MAX_ADMISSION_STARTS = Workflow_Operational_Limits::MAX_CONCURRENT_STARTS;
    private const MAX_ACTIVE_ITEMS = Workflow_Operational_Limits::MAX_ACTIVE_BULK_ITEMS;
    private const START_SPACING_SECONDS = 6;
    private const POLL_BATCH_LIMIT = 5;
    private const ITEM_TIMEOUT_SECONDS = 120;
    private const CAMPAIGN_ITEM_TIMEOUT_SECONDS = 900;
    private const STARTING_TIMEOUT_SECONDS = 120;
    private const PRODUCT_IMAGES_RUN_LOCK_SECONDS = 90;
    private const CAMPAIGN_RUN_LOCK_SECONDS = 900;
    private bool $start_blocked = false;

    private function __construct() {
        $this->repo = Bulk_Repo::instance();
        $this->universal_item_dispatcher = new Universal_Bulk_Item_Dispatcher();
    }

    private const RESUME_RAMP_CAPS = [ 2, 4 ];

    public static function instance(): Bulk_Run_Runner {
        if ( null === self::$instance ) {
            self::$instance = new self();
        }
        return self::$instance;
    }

    private function start_universal_item( array $run, array $item, string $item_owner ): bool {
        $run_id = (int) ( $run['id'] ?? 0 );
        $user_id = (int) ( $run['user_id'] ?? 0 );
        $item_id = (int) ( $item['id'] ?? 0 );
        $workflow = $this->get_run_workflow( $run );
        $prepared = $this->universal_item_dispatcher->prepare( $run, $item );
        if ( is_wp_error( $prepared ) ) {
            return $this->persist_universal_start_error( $item_id, $user_id, $workflow, $item_owner, $prepared, false );
        }

        $durable = $this->universal_item_dispatcher->inspect_durable_request( $user_id, $prepared );
        if ( is_wp_error( $durable ) ) {
            return $this->persist_universal_start_error( $item_id, $user_id, $workflow, $item_owner, $durable, true );
        }
        if ( ! empty( $durable['exists'] ) ) {
            return $this->reconcile_universal_durable_start( $run, $item, $item_owner, $prepared, $durable );
        }

        $current_run = $this->repo->get_run_by_id( $run_id, $user_id );
        if ( ! is_array( $current_run ) || 'running' !== (string) ( $current_run['status'] ?? '' ) ) {
            return $this->settle_universal_pre_dispatch_item( $run, $item, $item_owner, $current_run );
        }

        $first_attempt_at = trim( (string) ( $item['started_at'] ?? '' ) );
        if ( '' === $first_attempt_at ) {
            $first_attempt_at = current_time( 'mysql', true );
            if (
                ! $this->repo->admit_item_start(
                    $item_id,
                    $user_id,
                    $workflow,
                    (string) $prepared['requestId'],
                    $first_attempt_at,
                    $item_owner
                )
            ) {
                $current_run = $this->repo->get_run_by_id( $run_id, $user_id );
                if ( ! is_array( $current_run ) || 'running' !== (string) ( $current_run['status'] ?? '' ) ) {
                    return $this->settle_universal_pre_dispatch_item( $run, $item, $item_owner, $current_run );
                }
                return $this->persist_universal_start_error(
                    $item_id,
                    $user_id,
                    $workflow,
                    $item_owner,
                    new \WP_Error(
                        'fotogenic_ai_bulk_dispatch_marker_failed',
                        __( 'The Bulk dispatch marker could not be saved, so the item was not submitted.', 'fotogenic-ai' ),
                        [ 'status' => 503 ]
                    ),
                    true
                );
            }
            $stored = $this->repo->get_item_by_id( $item_id, $user_id );
            if (
                ! is_array( $stored )
                || ! hash_equals( $first_attempt_at, (string) ( $stored['started_at'] ?? '' ) )
                || ! hash_equals( (string) $prepared['requestId'], (string) ( $stored['request_id'] ?? '' ) )
            ) {
                return $this->persist_universal_start_error(
                    $item_id,
                    $user_id,
                    $workflow,
                    $item_owner,
                    new \WP_Error(
                        'fotogenic_ai_bulk_dispatch_marker_failed',
                        __( 'The Bulk dispatch marker could not be confirmed, so the item was not submitted.', 'fotogenic-ai' ),
                        [ 'status' => 503 ]
                    ),
                    true
                );
            }
            $item = $stored;
        }

        $current_run = $this->repo->get_run_by_id( $run_id, $user_id );
        if ( ! is_array( $current_run ) || 'running' !== (string) ( $current_run['status'] ?? '' ) ) {
            return $this->settle_universal_pre_dispatch_item( $run, $item, $item_owner, $current_run );
        }

        $result = $this->universal_item_dispatcher->start( $prepared, $run, $item );
        if ( is_wp_error( $result ) ) {
            $durable = $this->universal_item_dispatcher->inspect_durable_request( $user_id, $prepared );
            if ( is_array( $durable ) && ! empty( $durable['exists'] ) ) {
                return $this->reconcile_universal_durable_start( $run, $item, $item_owner, $prepared, $durable );
            }
			if ( 'fotogenic_ai_universal_data_unavailable' === $result->get_error_code() ) {
				return $this->pause_universal_data_gate( $run, $item, $item_owner );
			}
            if ( $this->is_credit_gate_error( (string) $result->get_error_code() ) ) {
                return $this->pause_universal_credit_gate( $run, $item, $item_owner, $result );
            }
            return $this->persist_universal_start_error( $item_id, $user_id, $workflow, $item_owner, $result, false );
        }

        if (
            ! empty( $result['requestId'] )
            && ! hash_equals( (string) $prepared['requestId'], (string) $result['requestId'] )
        ) {
            return $this->persist_universal_start_error(
                $item_id,
                $user_id,
                $workflow,
                $item_owner,
                new \WP_Error( 'fotogenic_ai_bulk_start_identity_mismatch', __( 'The Bulk start identity did not match.', 'fotogenic-ai' ), [ 'status' => 409 ] ),
                true
            );
        }

        $run_id_single = trim( (string) ( $result['runId'] ?? '' ) );
        $job_ids = $this->decode_job_ids( $result['jobIds'] ?? [] );
        if ( '' === $run_id_single || empty( $job_ids ) ) {
            $durable = $this->universal_item_dispatcher->inspect_durable_request( $user_id, $prepared );
            if ( is_array( $durable ) && ! empty( $durable['exists'] ) ) {
                return $this->reconcile_universal_durable_start( $run, $item, $item_owner, $prepared, $durable );
            }
            return $this->persist_universal_start_error(
                $item_id,
                $user_id,
                $workflow,
                $item_owner,
                new \WP_Error( 'fotogenic_ai_bulk_start_missing_ids', __( 'The Bulk start did not return tracking identifiers.', 'fotogenic-ai' ), [ 'status' => 500 ] ),
                false
            );
        }

        return $this->persist_universal_started_item( $item, $user_id, $item_owner, $prepared, $run_id_single, $job_ids );
    }

    private function settle_universal_pre_dispatch_item(
        array $run,
        array $item,
        string $item_owner,
        ?array $current_run
    ): bool {
        $status = sanitize_key( (string) ( $current_run['status'] ?? '' ) );
        $paused = 'paused' === $status;
        return $this->repo->update_claimed_item_if_status(
            (int) ( $item['id'] ?? 0 ),
            (int) ( $run['user_id'] ?? 0 ),
            $this->get_run_workflow( $run ),
            'starting',
            $item_owner,
            [
                'status'        => $paused ? 'pending' : 'skipped',
                'started_at'    => null,
                'error_code'    => $paused ? null : 'RUN_NOT_ACTIVE',
                'error_message' => $paused
                    ? null
                    : __( 'The Bulk item was not submitted because the run is no longer active.', 'fotogenic-ai' ),
                'finished_at'   => $paused ? null : current_time( 'mysql', true ),
            ]
        );
    }

    private function reconcile_universal_durable_start(
        array $run,
        array $item,
        string $item_owner,
        array $prepared,
        array $durable
    ): bool {
        $user_id = (int) ( $run['user_id'] ?? 0 );
        $item_id = (int) ( $item['id'] ?? 0 );
        $workflow = (string) $prepared['workflow'];
        $job_ids = $this->decode_job_ids( $durable['jobIds'] ?? [] );
        $run_id_single = trim( (string) ( $durable['runId'] ?? '' ) );
        if ( '' !== $run_id_single && ! empty( $job_ids ) ) {
            return $this->persist_universal_started_item( $item, $user_id, $item_owner, $prepared, $run_id_single, $job_ids );
        }

        $state = sanitize_key( (string) ( $durable['state'] ?? '' ) );
        if ( in_array( $state, Workflow_Request_Store::FAILURE_STATUSES, true ) ) {
            return $this->repo->update_claimed_item_if_status(
                $item_id,
                $user_id,
                $workflow,
                'starting',
                $item_owner,
                [
                    'status'        => 'failed',
                    'error_code'    => 'START_REJECTED',
                    'error_message' => __( 'The saved Bulk start reached a terminal failure.', 'fotogenic-ai' ),
                    'finished_at'   => current_time( 'mysql', true ),
                ]
            );
        }

        $started_at = trim( (string) ( $item['started_at'] ?? '' ) );
        $started_ts = '' !== $started_at ? strtotime( $started_at . ' UTC' ) : false;
        $expired = false !== $started_ts && ( time() - $started_ts ) >= $this->get_starting_timeout_seconds();
        if ( ! empty( $durable['uncertain'] ) || $expired ) {
            return $this->repo->update_claimed_item_if_status(
                $item_id,
                $user_id,
                $workflow,
                'starting',
                $item_owner,
                [
                    'status'        => 'uncertain',
                    'error_code'    => 'START_UNCERTAIN',
                    'error_message' => __( 'The Bulk start could not be confirmed and was not submitted again.', 'fotogenic-ai' ),
                    'finished_at'   => current_time( 'mysql', true ),
                ]
            );
        }

        return $this->repo->update_claimed_item_if_status(
            $item_id,
            $user_id,
            $workflow,
            'starting',
            $item_owner,
            [
                'status'        => 'starting',
                'error_code'    => 'START_RECONCILING',
                'error_message' => __( 'The Bulk start is being reconciled without resubmission.', 'fotogenic-ai' ),
            ]
        );
    }

    private function persist_universal_started_item(
        array $item,
        int $user_id,
        string $item_owner,
        array $prepared,
        string $run_id_single,
        array $job_ids
    ): bool {
        $item_id = (int) ( $item['id'] ?? 0 );
        $workflow = (string) $prepared['workflow'];
        $meta = $this->decode_filters_snapshot( $item['meta'] ?? '' );
        $meta['expected_output_count'] = (int) $prepared['expectedOutputCount'];
        $meta['source_attachment_ids'] = (array) $prepared['attachmentIds'];
        return $this->repo->update_claimed_item_if_status(
            $item_id,
            $user_id,
            $workflow,
            'starting',
            $item_owner,
            [
                'status'                => 'generating',
                'run_id_single'         => $run_id_single,
                'job_ids'               => wp_json_encode( $job_ids ),
                'generation_started_at' => current_time( 'mysql', true ),
                'error_code'            => null,
                'error_message'         => null,
                'meta'                  => wp_json_encode( $meta ),
            ]
        );
    }

    private function persist_universal_start_error(
        int $item_id,
        int $user_id,
        string $workflow,
        string $item_owner,
        \WP_Error $error,
        bool $uncertain
    ): bool {
        return $this->repo->update_claimed_item_if_status(
            $item_id,
            $user_id,
            $workflow,
            'starting',
            $item_owner,
            [
                'status'        => $uncertain ? 'uncertain' : 'failed',
                'error_code'    => $this->normalize_bulk_error_code( (string) $error->get_error_code() ),
                'error_message' => $uncertain
                    ? __( 'The Bulk start state could not be reconciled and was not submitted again.', 'fotogenic-ai' )
                    : __( 'The Bulk item could not be started.', 'fotogenic-ai' ),
                'finished_at'   => current_time( 'mysql', true ),
            ]
        );
    }

	private function pause_universal_credit_gate( array $run, array $item, string $item_owner, \WP_Error $error ): bool {
        $item_id = (int) ( $item['id'] ?? 0 );
        $user_id = (int) ( $run['user_id'] ?? 0 );
		$workflow = $this->get_run_workflow( $run );
		$insufficient = $this->is_insufficient_credit_error( (string) $error->get_error_code() );
		$authority_changed = $this->is_authority_gate_error( (string) $error->get_error_code() );
        $updated = $this->repo->update_claimed_item_if_status(
            $item_id,
            $user_id,
            $workflow,
            'starting',
            $item_owner,
            [
                'status'        => 'pending',
                'started_at'    => null,
				'error_code'    => $insufficient ? 'INSUFFICIENT_CREDITS' : ( $authority_changed ? 'AUTHORITY_CHANGED' : 'CREDIT_PREFLIGHT_UNAVAILABLE' ),
				'error_message' => $insufficient
					? __( 'Bulk is paused until enough credits are available.', 'fotogenic-ai' )
					: ( $authority_changed
						? __( 'Bulk is paused because the connected account, API environment, pricing, or configuration changed after Review.', 'fotogenic-ai' )
						: __( 'Bulk is paused because current pricing or credits could not be confirmed.', 'fotogenic-ai' ) ),
                'finished_at'   => null,
            ]
        );
        if ( $updated ) {
            $this->repo->update_run_if_status(
                (int) $run['id'],
                $user_id,
                'running',
                [
                    'status'        => 'paused',
					'status_reason' => $insufficient ? 'insufficient_credits' : ( $authority_changed ? 'authority_changed' : 'credit_preflight_unavailable' ),
                ],
                $workflow
            );
            $this->start_blocked = true;
        }
        return false;
    }

	private function pause_universal_data_gate( array $run, array $item, string $item_owner ): bool {
		$user_id  = (int) ( $run['user_id'] ?? 0 );
		$workflow = $this->get_run_workflow( $run );
		$updated  = $this->repo->update_claimed_item_if_status(
			(int) ( $item['id'] ?? 0 ),
			$user_id,
			$workflow,
			'starting',
			$item_owner,
			[
				'status'        => 'pending',
				'started_at'    => null,
				'error_code'    => 'UNIVERSAL_DATA_UNAVAILABLE',
				'error_message' => __( 'Bulk is paused until the Studio data upgrade is ready.', 'fotogenic-ai' ),
				'finished_at'   => null,
			]
		);
		if ( $updated ) {
			$this->repo->update_run_if_status(
				(int) ( $run['id'] ?? 0 ),
				$user_id,
				'running',
				[
					'status'        => 'paused',
					'status_reason' => 'universal_data_unavailable',
				],
				$workflow
			);
			$this->start_blocked = true;
		}
		return false;
	}

    private function poll_item_generation( array $run, array $item ): ?int {
        $run_id = (int) $run['id'];
        $user_id = (int) $run['user_id'];
        $item_id = (int) $item['id'];
        $workflow = $this->get_run_workflow( $run );
        $stored = $this->repo->get_item_by_id( $item_id, $user_id );
        if ( $stored ) {
            $item = $stored;
        }
        $started_at = $item['generation_started_at'] ?? ( $item['started_at'] ?? null );
        $run_id_single = (string) ( $item['run_id_single'] ?? '' );
        $job_ids = $this->decode_job_ids( $item['job_ids'] ?? '' );
        if ( '' === $run_id_single || empty( $job_ids ) ) {
            $this->repo->update_item( $item_id, $user_id, [
                'status' => 'failed',
                'error_code' => 'START_MISSING_IDS',
                'error_message' => 'Missing job identifiers.',
                'finished_at' => current_time( 'mysql', true ),
            ], $workflow );
            $this->finalize_run_if_done( $run_id, $user_id, '', $workflow );
            return null;
        }

        $poll_count = (int) ( $item['poll_count'] ?? 0 );
        $started_ts = $started_at ? strtotime( (string) $started_at . ' UTC' ) : 0;
        $timeout_seconds = $this->get_item_timeout_seconds( $workflow );
        if ( ( $started_ts && ( time() - $started_ts ) > $timeout_seconds ) || $poll_count > 200 ) {
            if ( $this->universal_item_dispatcher->supports( $item ) ) {
                $delay = $this->poll_universal_item( $run, $item );
                $fresh = $this->repo->get_item_by_id( $item_id, $user_id );
                if ( is_int( $delay ) && is_array( $fresh ) && 'generating' === (string) ( $fresh['status'] ?? '' ) ) {
                    $this->mark_item_finalization_pending( $fresh, $user_id, $workflow );
                }
                $this->finalize_run_if_done( $run_id, $user_id, '', $workflow );
                return null;
            }
			$this->mark_recovery_identity_unverifiable( $run, $item );
			$fresh = $this->repo->get_item_by_id( $item_id, $user_id );
			if ( is_array( $fresh ) && 'generating' === (string) ( $fresh['status'] ?? '' ) ) {
				$this->mark_item_finalization_pending( $fresh, $user_id, $workflow );
			}
            $this->finalize_run_if_done( $run_id, $user_id, '', $workflow );
            return null;
        }

        if ( $this->universal_item_dispatcher->supports( $item ) ) {
            return $this->poll_universal_item( $run, $item );
        }
		return $this->mark_recovery_identity_unverifiable( $run, $item );
    }

    private function poll_universal_item( array $run, array $item ): ?int {
        $item_id = (int) ( $item['id'] ?? 0 );
        $user_id = (int) ( $run['user_id'] ?? 0 );
        $workflow = $this->get_run_workflow( $run );
        $prepared = $this->universal_item_dispatcher->prepare( $run, $item );
        if ( is_wp_error( $prepared ) ) {
            $this->mark_universal_poll_failed( $item, $user_id, 'COMMAND_INVALID', __( 'The saved Bulk command is invalid.', 'fotogenic-ai' ), $workflow );
            return null;
        }
        $result = $this->universal_item_dispatcher->poll( $run, $item, $prepared );
        $poll_count = (int) ( $item['poll_count'] ?? 0 );
        if ( is_wp_error( $result ) ) {
            if ( $this->universal_item_dispatcher->poll_error_is_retryable( $result ) ) {
                $next_count = $poll_count + 1;
                $this->repo->update_item(
                    $item_id,
                    $user_id,
                    [
                        'last_polled_at' => current_time( 'mysql', true ),
                        'poll_count'     => $next_count,
                        'error_code'    => $this->normalize_bulk_error_code( (string) $result->get_error_code() ),
                        'error_message' => __( 'Bulk status is temporarily unavailable.', 'fotogenic-ai' ),
                    ],
                    $workflow
                );
                return $this->next_poll_delay( [ 'poll_count' => $next_count ] );
            }
            $this->mark_universal_poll_failed(
                $item,
                $user_id,
                $this->normalize_bulk_error_code( (string) $result->get_error_code() ),
                __( 'Bulk status could not be read.', 'fotogenic-ai' ),
                $workflow
            );
			return null;
		}

		$status = (string) ( $result['status'] ?? '' );
		if ( 'processing' === $status ) {
            if ( 'generating' !== (string) ( $item['status'] ?? '' ) ) {
                $this->mark_item_finalization_pending( $item, $user_id, $workflow );
                return null;
            }
            $next_count = $poll_count + 1;
            $this->repo->update_item(
                $item_id,
                $user_id,
                [
                    'last_polled_at' => current_time( 'mysql', true ),
                    'poll_count'     => $next_count,
                    'error_code'    => null,
                    'error_message' => null,
                ],
                $workflow
            );
            return $this->next_poll_delay( [ 'poll_count' => $next_count ] );
        }

        if ( in_array( $status, [ 'completed', 'partial' ], true ) ) {
            $attachment_ids = array_values( array_unique( array_filter( array_map( 'absint', (array) ( $result['attachmentIds'] ?? [] ) ) ) ) );
            if ( empty( $attachment_ids ) ) {
				if ( ! empty( $result['finalizationPending'] ) ) {
                    $this->mark_item_finalization_pending( $item, $user_id, $workflow );
                    return null;
                }
                $this->mark_universal_poll_failed( $item, $user_id, 'MISSING_ATTACHMENTS', __( 'Bulk completed without saved Media Library images.', 'fotogenic-ai' ), $workflow );
                return null;
            }
            $expected = (int) $prepared['expectedOutputCount'];
			$finalization_pending = ! empty( $result['finalizationPending'] );
			$partial = $finalization_pending || ! empty( $result['partial'] ) || count( $attachment_ids ) !== $expected;
            $meta = $this->decode_filters_snapshot( $item['meta'] ?? '' );
            $meta['saved_output_count'] = count( $attachment_ids );
            $meta['partial'] = $partial;
			if ( $finalization_pending ) {
				$meta['finalization_pending'] = true;
				$this->repo->update_item(
					$item_id,
					$user_id,
					[
						'status'         => 'generating',
						'attachment_id'  => (int) $attachment_ids[0],
						'attachment_ids' => wp_json_encode( $attachment_ids ),
						'last_polled_at' => current_time( 'mysql', true ),
						'poll_count'     => $poll_count + 1,
						'error_code'     => 'FINALIZATION_PENDING',
						'error_message'  => __( 'Accepted outputs are still being finalized.', 'fotogenic-ai' ),
						'finished_at'    => null,
						'meta'           => wp_json_encode( $meta ),
					],
					$workflow
				);
				return $this->next_poll_delay( [ 'poll_count' => $poll_count + 1 ] );
            }
			unset( $meta['finalization_pending'] );
            $this->repo->update_item(
                $item_id,
                $user_id,
                [
                    'status'         => $partial ? 'partial' : 'completed',
                    'attachment_id'  => (int) $attachment_ids[0],
                    'attachment_ids' => wp_json_encode( $attachment_ids ),
					'error_code'     => $partial ? 'PARTIAL' : null,
					'error_message'  => $partial ? __( 'Bulk completed with a different output count than requested.', 'fotogenic-ai' ) : null,
                    'finished_at'    => current_time( 'mysql', true ),
                    'meta'           => wp_json_encode( $meta ),
                ],
                $workflow
            );
            return null;
        }

		if ( ! empty( $result['finalizationPending'] ) ) {
            $this->mark_item_finalization_pending( $item, $user_id, $workflow );
            return null;
        }

        $this->mark_universal_poll_failed(
            $item,
            $user_id,
            $this->normalize_bulk_error_code( (string) ( $result['errorCode'] ?? ( 'timeout' === $status ? 'TIMEOUT' : 'FAILED' ) ) ),
            __( 'Bulk generation did not complete.', 'fotogenic-ai' ),
            $workflow
        );
        return null;
    }

	private function mark_recovery_identity_unverifiable( array $run, array $item ): ?int {
		$this->repo->update_item(
			(int) ( $item['id'] ?? 0 ),
			(int) ( $run['user_id'] ?? 0 ),
			[
				'status'        => 'uncertain',
				'error_code'    => 'RECOVERY_IDENTITY_UNVERIFIABLE',
				'error_message' => __( 'The accepted generation identity could not be verified for Bulk polling.', 'fotogenic-ai' ),
				'finished_at'   => current_time( 'mysql', true ),
			],
			$this->get_run_workflow( $run )
		);
		return null;
	}

    private function mark_item_failed( int $item_id, int $user_id, string $error_code, string $message, string $workflow = '' ): void {
        $this->repo->update_item( $item_id, $user_id, [
            'status' => 'failed',
            'error_code' => $error_code,
            'error_message' => $message,
            'finished_at' => current_time( 'mysql', true ),
        ], $workflow );
    }

    private function mark_universal_poll_failed( array $item, int $user_id, string $error_code, string $message, string $workflow ): void {
        $item_id = (int) ( $item['id'] ?? 0 );
        if ( $item_id <= 0 || $user_id <= 0 ) {
            return;
        }
        $attachment_ids = $this->get_item_attachment_ids( $item );
        $meta = $this->decode_filters_snapshot( $item['meta'] ?? '' );
        unset( $meta['finalization_pending'] );
        if ( ! empty( $attachment_ids ) ) {
            $meta['saved_output_count'] = count( $attachment_ids );
            $meta['partial'] = true;
        }
        $this->repo->update_item(
            $item_id,
            $user_id,
            [
                'status'        => empty( $attachment_ids ) ? 'failed' : 'partial',
                'error_code'    => $this->normalize_bulk_error_code( $error_code ),
                'error_message' => $message,
                'finished_at'   => current_time( 'mysql', true ),
                'meta'          => wp_json_encode( $meta ),
            ],
            $workflow
        );
    }

    private function mark_item_finalization_pending( array $item, int $user_id, string $workflow ): void {
        $item_id = (int) ( $item['id'] ?? 0 );
        if ( $item_id <= 0 || $user_id <= 0 ) {
            return;
        }
        $meta = $this->decode_filters_snapshot( $item['meta'] ?? '' );
        $meta['finalization_pending'] = true;
        $attachment_ids = $this->get_item_attachment_ids( $item );
        $this->repo->update_item(
            $item_id,
            $user_id,
            [
				'status'        => 'generating',
				'attachment_id' => (int) ( $attachment_ids[0] ?? 0 ),
				'attachment_ids'=> wp_json_encode( $attachment_ids ),
                'error_code'    => 'FINALIZATION_PENDING',
				'error_message' => __( 'Accepted generation is still being finalized.', 'fotogenic-ai' ),
				'finished_at'   => null,
                'meta'          => wp_json_encode( $meta ),
            ],
            $workflow
        );
    }

    private function get_item_attachment_ids( array $item ): array {
        $attachment_ids   = $this->decode_job_ids( $item['attachment_ids'] ?? '' );
        $attachment_ids[] = absint( $item['attachment_id'] ?? 0 );
        return array_values( array_unique( array_filter( $attachment_ids ) ) );
    }

    private function refresh_terminal_run_summary( int $run_id, int $user_id, string $workflow ): void {
        $run = $this->repo->get_run_by_id( $run_id, $user_id );
        if ( empty( $run ) ) {
            return;
        }
        $status = sanitize_key( (string) ( $run['status'] ?? '' ) );
        if ( ! in_array( $status, [ 'completed', 'stopped', 'failed' ], true ) ) {
            return;
        }
        $counts = $this->repo->count_items_by_status( $run_id, $user_id );
        $completed = (int) ( $counts['completed'] ?? 0 );
        $partial = (int) ( $counts['partial'] ?? 0 );
        $failed = (int) ( $counts['failed'] ?? 0 );
        $skipped = (int) ( $counts['skipped'] ?? 0 );
        $uncertain = (int) ( $counts['uncertain'] ?? 0 );
        $status_reason = 'stopped' === $status
            ? 'user_stop'
            : ( $uncertain > 0 ? 'uncertain_items' : ( $partial + $failed + $skipped > 0 ? 'partial' : '' ) );
        $this->repo->update_run_if_status(
            $run_id,
            $user_id,
            $status,
            [
                'status'          => $status,
                'status_reason'   => $status_reason,
                'completed_count' => $completed + $partial,
                'failed_count'    => $failed + $skipped + $uncertain,
            ],
            $workflow
        );
    }

    private function finalize_run_if_done( int $run_id, int $user_id, string $status = '', string $workflow = '' ): bool {
        if ( ! $this->repo->acquire_admission_lock( $user_id ) ) {
            return false;
        }
        try {
            $run = $this->repo->get_run_by_id( $run_id, $user_id );
            if ( empty( $run ) ) {
                return true;
            }
            $status = sanitize_key( (string) ( $run['status'] ?? '' ) );
            if ( ! in_array( $status, [ 'running', 'stopping' ], true ) ) {
                return true;
            }
            $workflow = $this->get_run_workflow( $run );
            $counts = $this->repo->count_items_by_status( $run_id, $user_id );
            $pending = (int) ( $counts['pending'] ?? 0 );
            $starting = (int) ( $counts['starting'] ?? 0 );
            $generating = (int) ( $counts['generating'] ?? 0 );
            if ( 0 !== $pending || 0 !== $starting || 0 !== $generating ) {
                return true;
            }
            $completed = (int) ( $counts['completed'] ?? 0 );
            $partial = (int) ( $counts['partial'] ?? 0 );
            $failed = (int) ( $counts['failed'] ?? 0 );
            $skipped = (int) ( $counts['skipped'] ?? 0 );
            $uncertain = (int) ( $counts['uncertain'] ?? 0 );
            $final_status = ( 'stopping' === $status ) ? 'stopped' : 'completed';
            $status_reason = 'stopped' === $final_status
                ? 'user_stop'
                : ( $uncertain > 0 ? 'uncertain_items' : ( $partial + $failed + $skipped > 0 ? 'partial' : '' ) );
            $this->repo->update_run_if_status( $run_id, $user_id, $status, [
                'status'          => $final_status,
                'status_reason'   => $status_reason,
                'finished_at'     => current_time( 'mysql', true ),
                'completed_count' => $completed + $partial,
                'failed_count'    => $failed + $skipped + $uncertain,
            ], $workflow );
            return true;
        } finally {
            $this->repo->release_admission_lock( $user_id );
        }
    }

    private function schedule_next_tick( array $run, int $delay ): bool {
        $run_id = (int) ( $run['id'] ?? 0 );
        $run_uuid = (string) ( $run['run_uuid'] ?? '' );
        $user_id = (int) ( $run['user_id'] ?? 0 );
        $delay = max( 1, $delay );
        if ( Bulk_Scheduler::instance()->schedule_tick( $run_uuid, $user_id, $delay ) ) {
            return true;
        }

        $status = sanitize_key( (string) ( $run['status'] ?? '' ) );
        $changes = [
            'status_reason' => 'scheduler_unavailable',
            'last_error'    => 'Bulk continuation could not be scheduled.',
        ];
        if ( 'running' === $status ) {
            $changes['status'] = 'paused';
        }
        $this->repo->update_run_if_status( $run_id, $user_id, $status, $changes, $this->get_run_workflow( $run ) );
        if ( Logger::instance()->enabled() ) {
            Logger::instance()->warn( 'bulk_scheduler_unavailable', [
                'run_id'        => 'bulk:' . $run_uuid,
                'bulk_run_uuid' => $run_uuid,
                'user_id'       => $user_id,
                'workflow'      => $this->get_run_workflow( $run ),
            ] );
        }
        return false;
    }

    private function get_start_spacing_delay( string $run_uuid, int $now ): int {
        $last = $this->get_last_start_ts( $run_uuid );
        if ( $last <= 0 ) {
            return 0;
        }
        $elapsed = $now - $last;
        if ( $elapsed >= self::START_SPACING_SECONDS ) {
            return 0;
        }
        return self::START_SPACING_SECONDS - $elapsed;
    }

    private function get_last_start_ts( string $run_uuid ): int {
        $value = get_transient( $this->get_start_spacing_key( $run_uuid ) );
        return is_numeric( $value ) ? (int) $value : 0;
    }

    private function set_last_start_ts( string $run_uuid, int $timestamp = 0 ): void {
        $value = $timestamp > 0 ? $timestamp : time();
        set_transient( $this->get_start_spacing_key( $run_uuid ), $value, 3600 );
    }

    private function get_start_spacing_key( string $run_uuid ): string {
        return 'fotogenic_ai_bulk_last_start_' . $run_uuid;
    }

    private function get_resume_ramp_key( string $run_uuid ): string {
        return 'fotogenic_ai_bulk_resume_ramp_' . $run_uuid;
    }

    private function advance_resume_ramp_if_used( string $run_uuid ): void {
        $phase = get_transient( $this->get_resume_ramp_key( $run_uuid ) );
        if ( false === $phase || ! is_numeric( $phase ) ) {
            return;
        }
        $next = (int) $phase + 1;
        if ( $next <= count( self::RESUME_RAMP_CAPS ) ) {
            set_transient( $this->get_resume_ramp_key( $run_uuid ), $next, 600 );
        } else {
            delete_transient( $this->get_resume_ramp_key( $run_uuid ) );
        }
    }

    private function get_start_slots( string $run_uuid, int $now ): int {
        $last = $this->get_last_start_ts( $run_uuid );
        $ramp_phase = get_transient( $this->get_resume_ramp_key( $run_uuid ) );
        $in_resume_ramp = false !== $ramp_phase && is_numeric( $ramp_phase );
        $phase = $in_resume_ramp ? (int) $ramp_phase : -1;

		$slots = self::MAX_ADMISSION_STARTS;
        if ( $last > 0 ) {
            $elapsed = $now - $last;
            if ( $elapsed < self::START_SPACING_SECONDS ) {
                return 0;
            }
            $slots = (int) floor( $elapsed / self::START_SPACING_SECONDS );
        } elseif ( $in_resume_ramp && $phase < count( self::RESUME_RAMP_CAPS ) ) {
            $slots = self::RESUME_RAMP_CAPS[ $phase ];
        }

        if ( $in_resume_ramp ) {
            $cap = $phase < count( self::RESUME_RAMP_CAPS )
                ? min( self::RESUME_RAMP_CAPS[ $phase ], self::MAX_ADMISSION_STARTS )
                : self::MAX_ADMISSION_STARTS;
            return min( $slots, $cap );
        }
        return min( $slots, self::MAX_ADMISSION_STARTS );
    }

    private function get_available_active_item_capacity( int $starting, int $generating ): int {
        $active_items = max( 0, $starting ) + max( 0, $generating );
        return max( 0, self::MAX_ACTIVE_ITEMS - $active_items );
    }

    private function get_poll_delay_seconds( array $item, int $now ): int {
        $last = isset( $item['last_polled_at'] ) ? (string) $item['last_polled_at'] : '';
        $poll_delay = $this->next_poll_delay( $item );
        if ( '' === $last ) {
            return 0;
        }
        $last_ts = strtotime( $last . ' UTC' );
        if ( ! $last_ts ) {
            return 0;
        }
        $elapsed = $now - $last_ts;
        if ( $elapsed >= $poll_delay ) {
            return 0;
        }
        return $poll_delay - $elapsed;
    }

    private function decode_job_ids( $job_ids ): array {
        if ( is_array( $job_ids ) ) {
            return $job_ids;
        }
        if ( is_string( $job_ids ) && $job_ids !== '' ) {
            $decoded = json_decode( $job_ids, true );
            if ( json_last_error() === JSON_ERROR_NONE && is_array( $decoded ) ) {
                return $decoded;
            }
            return [ $job_ids ];
        }
        return [];
    }

    private function get_item_timeout_seconds( string $workflow ): int {
        if ( 'campaign_set' === $workflow ) {
            $seconds = (int) apply_filters( 'fotogenic_ai_campaign_bulk_item_timeout_seconds', self::CAMPAIGN_ITEM_TIMEOUT_SECONDS );
            return max( self::CAMPAIGN_ITEM_TIMEOUT_SECONDS, $seconds );
        }
        $seconds = (int) apply_filters( 'fotogenic_ai_bulk_item_timeout_seconds', self::ITEM_TIMEOUT_SECONDS );
        return max( 60, $seconds );
    }

    private function get_starting_timeout_seconds(): int {
        $seconds = (int) apply_filters( 'fotogenic_ai_bulk_starting_timeout_seconds', self::STARTING_TIMEOUT_SECONDS );
        return max( 60, $seconds );
    }

    private function fail_stuck_starting_items( array $run, int $now ): int {
        $run_id = (int) ( $run['id'] ?? 0 );
        $user_id = (int) ( $run['user_id'] ?? 0 );
        if ( $run_id <= 0 || $user_id <= 0 ) {
            return 0;
        }
        $workflow = $this->get_run_workflow( $run );
        $starting_items = $this->repo->list_items_by_statuses( $run_id, $user_id, [ 'starting' ], 50, 'updated_at', 'ASC' );
        if ( empty( $starting_items ) ) {
            return 0;
        }
        $timeout_seconds = $this->get_starting_timeout_seconds();
        $timed_out = 0;
        foreach ( $starting_items as $item ) {
            $item_id = (int) ( $item['id'] ?? 0 );
            if ( $item_id <= 0 ) {
                continue;
            }
            $updated_at = trim( (string) ( $item['updated_at'] ?? '' ) );
            $updated_ts = '' !== $updated_at ? strtotime( $updated_at . ' UTC' ) : false;
            $reconciliation_due = false === $updated_ts || ( $now - $updated_ts ) > $timeout_seconds;
            if ( '' !== trim( (string) ( $item['command_snapshot'] ?? '' ) ) ) {
                if ( ! $reconciliation_due ) {
                    continue;
                }
                if ( $this->reconcile_stuck_universal_item( $run, $item ) ) {
                    ++$timed_out;
                }
                continue;
            }
            if ( ! $reconciliation_due ) {
                continue;
            }
            if ( $this->fail_invalid_stuck_item( $run, $item ) ) {
                ++$timed_out;
            }
        }
        if ( $timed_out > 0 && Logger::instance()->enabled() ) {
            Logger::instance()->warn( 'bulk_start_timeout', [
                'run_id' => 'bulk:' . (string) ( $run['run_uuid'] ?? '' ),
                'bulk_run_uuid' => (string) ( $run['run_uuid'] ?? '' ),
                'timed_out_count' => $timed_out,
                'reconciliation_timeout_seconds' => $timeout_seconds,
                'campaign_timeout_seconds' => $this->get_item_timeout_seconds( 'campaign_set' ),
            ] );
        }
        return $timed_out;
    }

	private function starting_item_belongs_to_run( array $run, array $item ): bool {
        return (int) ( $item['run_id'] ?? 0 ) === (int) ( $run['id'] ?? 0 );
    }

    private function get_run_workflow( array $run ): string {
        $workflow = sanitize_key( (string) ( $run['workflow'] ?? '' ) );
        return in_array( $workflow, [ 'product_images', 'campaign_set', 'increase_resolution' ], true )
            ? $workflow
	            : '';
    }

    private function get_run_lock_seconds( array $run ): int {
        return 'campaign_set' === $this->get_run_workflow( $run )
            ? self::CAMPAIGN_RUN_LOCK_SECONDS
            : self::PRODUCT_IMAGES_RUN_LOCK_SECONDS;
    }

    private function is_insufficient_credit_error( string $error_code ): bool {
        $error_code = strtoupper( str_replace( '-', '_', sanitize_key( $error_code ) ) );
        return fotogenic_ai_str_contains( $error_code, 'INSUFFICIENT_CREDIT' )
            || fotogenic_ai_str_contains( $error_code, 'PAYMENT_REQUIRED' )
            || 'HTTP_402' === $error_code
            || '402' === $error_code;
    }

	private function is_credit_gate_error( string $error_code ): bool {
        $error_code = strtoupper( str_replace( '-', '_', sanitize_key( $error_code ) ) );
		return $this->is_insufficient_credit_error( $error_code )
			|| $this->is_authority_gate_error( $error_code )
			|| fotogenic_ai_str_contains( $error_code, 'CREDIT_PREFLIGHT_UNAVAILABLE' )
            || fotogenic_ai_str_contains( $error_code, 'PRICING_PREFLIGHT_UNAVAILABLE' )
            || fotogenic_ai_str_contains( $error_code, 'PREFLIGHT_UNAVAILABLE' )
            || fotogenic_ai_str_contains( $error_code, 'PRICING_UNAVAILABLE' )
			|| fotogenic_ai_str_contains( $error_code, 'CREDITS_UNAVAILABLE' );
	}

	private function is_authority_gate_error( string $error_code ): bool {
		$error_code = strtoupper( str_replace( '-', '_', sanitize_key( $error_code ) ) );
		return fotogenic_ai_str_contains( $error_code, 'AUTHORITY_CHANGED' );
	}

    private function fail_invalid_stuck_item( array $run, array $item ): bool {
        $user_id = (int) ( $run['user_id'] ?? 0 );
        $workflow = $this->get_run_workflow( $run );
        $updated = $this->repo->update_item_if_status(
            (int) ( $item['id'] ?? 0 ),
            $user_id,
            'starting',
            [
                'status'        => 'failed',
                'error_code'    => 'COMMAND_INVALID',
                'error_message' => __( 'The saved Bulk command is invalid.', 'fotogenic-ai' ),
                'finished_at'   => current_time( 'mysql', true ),
            ],
            $workflow
        );
        return $updated;
    }

    private function reconcile_stuck_universal_item( array $run, array $item ): bool {
        $item_id = (int) ( $item['id'] ?? 0 );
        $user_id = (int) ( $run['user_id'] ?? 0 );
        $workflow = $this->get_run_workflow( $run );
        $prepared = $this->universal_item_dispatcher->prepare( $run, $item );
        if ( is_wp_error( $prepared ) ) {
            return $this->repo->update_item_if_status(
                $item_id,
                $user_id,
                'starting',
                [
                    'status'        => 'failed',
                    'error_code'    => 'COMMAND_INVALID',
                    'error_message' => __( 'The saved Bulk command is invalid.', 'fotogenic-ai' ),
                    'finished_at'   => current_time( 'mysql', true ),
                ],
                $workflow
            );
        }
        $durable = $this->universal_item_dispatcher->inspect_durable_request( $user_id, $prepared );
        if ( is_wp_error( $durable ) ) {
            return $this->repo->update_item_if_status(
                $item_id,
                $user_id,
                'starting',
                [
                    'status'        => 'uncertain',
                    'error_code'    => 'START_RECONCILIATION_CONFLICT',
                    'error_message' => __( 'The Bulk start identity could not be reconciled and was not submitted again.', 'fotogenic-ai' ),
                    'finished_at'   => current_time( 'mysql', true ),
                ],
                $workflow
            );
        }
        if ( empty( $durable['exists'] ) ) {
            $run_status = (string) ( $run['status'] ?? '' );
            $recoverable = in_array( $run_status, [ 'running', 'paused' ], true );
            return $this->repo->update_item_if_status(
                $item_id,
                $user_id,
                'starting',
                [
                    'status'        => $recoverable ? 'pending' : 'skipped',
                    'started_at'    => null,
                    'error_code'    => $recoverable ? 'SAFE_PRE_DISPATCH_RECOVERY' : 'STOPPED_BY_USER',
                    'error_message' => $recoverable
                        ? __( 'The item was safely restored before any billable admission.', 'fotogenic-ai' )
                        : __( 'The item was stopped before any billable admission.', 'fotogenic-ai' ),
                    'finished_at'   => $recoverable ? null : current_time( 'mysql', true ),
                ],
                $workflow
            );
        }
        $job_ids = $this->decode_job_ids( $durable['jobIds'] ?? [] );
        $run_id_single = trim( (string) ( $durable['runId'] ?? '' ) );
        if ( '' !== $run_id_single && ! empty( $job_ids ) ) {
            $meta = $this->decode_filters_snapshot( $item['meta'] ?? '' );
            $meta['expected_output_count'] = (int) $prepared['expectedOutputCount'];
            $meta['source_attachment_ids'] = (array) $prepared['attachmentIds'];
            return $this->repo->update_item_if_status(
                $item_id,
                $user_id,
                'starting',
                [
                    'status'                => 'generating',
                    'run_id_single'         => $run_id_single,
                    'job_ids'               => wp_json_encode( $job_ids ),
                    'generation_started_at' => current_time( 'mysql', true ),
                    'error_code'            => null,
                    'error_message'         => null,
                    'meta'                  => wp_json_encode( $meta ),
                ],
                $workflow
            );
        }
        $state = sanitize_key( (string) ( $durable['state'] ?? '' ) );
        $failed = in_array( $state, Workflow_Request_Store::FAILURE_STATUSES, true );
        return $this->repo->update_item_if_status(
            $item_id,
            $user_id,
            'starting',
            [
                'status'        => $failed ? 'failed' : 'uncertain',
                'error_code'    => $failed ? 'START_REJECTED' : 'START_UNCERTAIN',
                'error_message' => $failed
                    ? __( 'The saved Bulk start reached a terminal failure.', 'fotogenic-ai' )
                    : __( 'The Bulk start could not be confirmed and was not submitted again.', 'fotogenic-ai' ),
                'finished_at'   => current_time( 'mysql', true ),
            ],
            $workflow
        );
    }

    private function normalize_bulk_error_code( string $code ): string {
        $code = strtoupper( str_replace( '-', '_', sanitize_key( $code ) ) );
        return '' !== $code ? substr( $code, 0, 64 ) : 'BULK_FAILED';
    }

    private function next_poll_delay( array $item ): int {
        $poll_count = (int) ( $item['poll_count'] ?? 0 );
        $steps = [ 5, 10, 15, 20, 25, 30 ];
        $index = min( count( $steps ) - 1, max( 0, $poll_count ) );
        return $steps[ $index ];
    }

    private function decode_filters_snapshot( $raw ): array {
        if ( is_string( $raw ) && $raw !== '' ) {
            $decoded = json_decode( $raw, true );
            if ( json_last_error() === JSON_ERROR_NONE && is_array( $decoded ) ) {
                return $decoded;
            }
        }
        return is_array( $raw ) ? $raw : [];
    }

    /**
     * Reset start spacing and set resume ramp so items start gradually (2, 4, 5) after pause/resume.
     */
    public function reset_resume_ramp( string $run_uuid ): void {
        delete_transient( $this->get_start_spacing_key( $run_uuid ) );
        set_transient( $this->get_resume_ramp_key( $run_uuid ), 0, 600 );
    }

	/**
	 * Synchronize one finalized Logger run back to its exact Bulk item.
	 *
	 * This is a non-billable projection. It accepts only the immutable identity
	 * already shared by the Logger run, Bulk run, Bulk item, and attachments.
	 */
	public function sync_finalized_item( string $logger_run_id, string $final_status ): void {
		$logger_run_id = trim( $logger_run_id );
		$final_status  = sanitize_key( $final_status );
		if ( '' === $logger_run_id || ! in_array( $final_status, [ 'completed', 'partial', 'failed' ], true ) ) {
			return;
		}

		$row           = Logger::instance()->get_run_row( $logger_run_id );
		$run_uuid      = strtolower( trim( (string) ( $row['bulk_run_uuid'] ?? '' ) ) );
		$item_id       = (int) ( $row['bulk_item_id'] ?? 0 );
		$user_id       = (int) ( $row['user_id'] ?? 0 );
		$workflow      = sanitize_key( (string) ( $row['workflow'] ?? '' ) );
		$request_id    = strtolower( trim( (string) ( $row['request_id'] ?? '' ) ) );
		$run_item_uuid = strtolower( trim( (string) ( $row['run_item_uuid'] ?? '' ) ) );
		if (
			! wp_is_uuid( $run_uuid )
			|| $item_id <= 0
			|| $user_id <= 0
			|| ! in_array( $workflow, [ 'product_images', 'campaign_set', 'increase_resolution' ], true )
			|| ! wp_is_uuid( $request_id )
			|| ! wp_is_uuid( $run_item_uuid )
		) {
			return;
		}

		$run  = $this->repo->get_run_by_uuid( $run_uuid, $user_id );
		$item = $this->repo->get_item_by_id( $item_id, $user_id );
		if ( ! is_array( $run ) || ! is_array( $item ) ) {
			return;
		}
		$logger_target = $this->normalize_persisted_target( $row['target_snapshot'] ?? null );
		$item_target   = $this->normalize_persisted_target( $item['target_snapshot'] ?? null );
		$logger_jobs   = $this->normalize_string_ids( $row['job_ids'] ?? $row['job_id'] ?? null );
		$item_jobs     = $this->normalize_string_ids( $item['job_ids'] ?? null );
		$campaign_group_uuid = strtolower( trim( (string) ( $row['campaign_group_uuid'] ?? '' ) ) );
		if (
			$final_status !== sanitize_key( (string) ( $row['final_status'] ?? '' ) )
			|| ! hash_equals( $logger_run_id, trim( (string) ( $item['run_id_single'] ?? '' ) ) )
			|| (int) ( $item['run_id'] ?? 0 ) !== (int) ( $run['id'] ?? 0 )
			|| (int) ( $run['user_id'] ?? 0 ) !== $user_id
			|| $workflow !== sanitize_key( (string) ( $run['workflow'] ?? '' ) )
			|| ! hash_equals( $run_item_uuid, strtolower( trim( (string) ( $item['run_item_uuid'] ?? '' ) ) ) )
			|| (int) ( $row['item_ordinal'] ?? 0 ) !== (int) ( $item['item_ordinal'] ?? 0 )
			|| ! hash_equals( $request_id, strtolower( trim( (string) ( $item['request_id'] ?? '' ) ) ) )
			|| ! hash_equals( $campaign_group_uuid, strtolower( trim( (string) ( $item['campaign_group_uuid'] ?? '' ) ) ) )
			|| sanitize_key( (string) ( $row['run_target'] ?? '' ) ) !== sanitize_key( (string) ( $item['subject_type'] ?? '' ) )
			|| (int) ( $row['subject_id'] ?? 0 ) !== (int) ( $item['subject_id'] ?? 0 )
			|| null === $logger_target
			|| $logger_target !== $item_target
			|| empty( $logger_jobs )
			|| $logger_jobs !== $item_jobs
		) {
			return;
		}

		$run_target = sanitize_key( (string) ( $row['run_target'] ?? '' ) );
		$subject_id = (int) ( $row['subject_id'] ?? 0 );
		$item_ordinal = (int) ( $row['item_ordinal'] ?? 0 );
		$attachment_map = $this->logger_attachment_map( $row, $logger_jobs );
		$attachment_ids = [];
		foreach ( $attachment_map as $output_key => $attachment_id ) {
			$job_id = Generate_Output_Map::job_id_for_key( (string) $output_key, $logger_jobs );
			$campaign_meta_complete = metadata_exists( 'post', $attachment_id, '_fotogenic_ai_asset_id' )
				&& metadata_exists( 'post', $attachment_id, '_fotogenic_ai_asset_order' );
			$output_identity_matches = 'campaign_set' !== $workflow
				? hash_equals( $job_id, (string) $output_key )
				: $campaign_meta_complete && hash_equals(
					(string) $output_key,
					Generate_Output_Map::campaign_key(
						$job_id,
						(string) get_post_meta( $attachment_id, '_fotogenic_ai_asset_id', true ),
						(int) get_post_meta( $attachment_id, '_fotogenic_ai_asset_order', true )
					)
				);
			$attachment_target = Workflow_Item_Identity::normalize_target_snapshot(
				get_post_meta( $attachment_id, '_fotogenic_ai_target_snapshot', true )
			);
			if (
				'' !== $job_id
				&& $output_identity_matches
				&&
				get_post( $attachment_id ) instanceof \WP_Post
				&& $user_id === (int) get_post_meta( $attachment_id, '_fotogenic_ai_user_id', true )
				&& hash_equals( $job_id, (string) get_post_meta( $attachment_id, '_fotogenic_ai_job_id', true ) )
				&& hash_equals( $workflow, sanitize_key( (string) get_post_meta( $attachment_id, '_fotogenic_ai_workflow', true ) ) )
				&& hash_equals( $logger_run_id, (string) get_post_meta( $attachment_id, '_fotogenic_ai_run_id', true ) )
				&& hash_equals( $run_item_uuid, strtolower( (string) get_post_meta( $attachment_id, '_fotogenic_ai_run_item_uuid', true ) ) )
				&& hash_equals( $request_id, strtolower( (string) get_post_meta( $attachment_id, '_fotogenic_ai_request_id', true ) ) )
				&& hash_equals( $run_target, sanitize_key( (string) get_post_meta( $attachment_id, '_fotogenic_ai_run_target', true ) ) )
				&& $subject_id === (int) get_post_meta( $attachment_id, '_fotogenic_ai_subject_id', true )
				&& hash_equals( $campaign_group_uuid, strtolower( (string) get_post_meta( $attachment_id, '_fotogenic_ai_campaign_group_uuid', true ) ) )
				&& $item_ordinal === (int) get_post_meta( $attachment_id, '_fotogenic_ai_item_ordinal', true )
				&& ! is_wp_error( $attachment_target )
				&& $logger_target === $attachment_target
			) {
				$attachment_ids[] = $attachment_id;
			}
		}
		$expected = max( 1, (int) ( $row['expected_output_count'] ?? count( $logger_jobs ) ) );
		$status   = count( $attachment_ids ) === $expected
			? 'completed'
			: ( ! empty( $attachment_ids ) ? 'partial' : 'failed' );
		$meta = $this->decode_filters_snapshot( $item['meta'] ?? '' );
		unset( $meta['finalization_pending'] );
		$meta['saved_output_count'] = count( $attachment_ids );
		$meta['partial']            = 'partial' === $status;
		$fields = [
			'status'         => $status,
			'attachment_id'  => (int) ( $attachment_ids[0] ?? 0 ),
			'attachment_ids' => wp_json_encode( $attachment_ids ),
			'error_code'     => 'completed' === $status ? null : ( 'partial' === $status ? 'PARTIAL' : 'FINALIZATION_FAILED' ),
			'error_message'  => 'completed' === $status
				? null
				: ( 'partial' === $status ? __( 'Some generated images could not be saved.', 'fotogenic-ai' ) : __( 'Generated images could not be finalized.', 'fotogenic-ai' ) ),
			'finished_at'    => current_time( 'mysql', true ),
			'meta'           => wp_json_encode( $meta ),
		];

		foreach ( [ 'generating', 'uncertain', 'partial', 'failed' ] as $expected_status ) {
			if ( $expected_status !== (string) ( $item['status'] ?? '' ) ) {
				continue;
			}
			if ( ! $this->repo->update_item_if_status( $item_id, $user_id, $expected_status, $fields, $workflow ) ) {
				return;
			}
			$this->finalize_run_if_done( (int) $run['id'], $user_id, '', $workflow );
			$this->refresh_terminal_run_summary( (int) $run['id'], $user_id, $workflow );
			return;
		}
	}

	/** @param mixed $raw Raw value. */
	private function normalize_persisted_target( $raw ): ?array {
		if ( is_string( $raw ) ) {
			$decoded = json_decode( $raw, true );
			$raw = is_array( $decoded ) ? $decoded : null;
		}
		$target = Workflow_Item_Identity::normalize_target_snapshot( $raw );
		return is_wp_error( $target ) ? null : $target;
	}

	/**
	 * @param mixed $raw Raw value.
	 * @return string[]
	 */
	private function normalize_string_ids( $raw ): array {
		$values = $this->decode_job_ids( $raw );
		$ids    = [];
		foreach ( $values as $value ) {
			$value = trim( sanitize_text_field( (string) $value ) );
			if ( '' !== $value && strlen( $value ) <= 64 && ! in_array( $value, $ids, true ) ) {
				$ids[] = $value;
			}
		}
		return $ids;
	}

	/** @return int[] */
	private function logger_attachment_map( array $row, array $job_ids ): array {
		$map = $row['attachments_map'] ?? [];
		if ( is_string( $map ) ) {
			$decoded = json_decode( $map, true );
			$map = is_array( $decoded ) ? $decoded : [];
		}
		return Generate_Output_Map::normalize( $map, $job_ids, sanitize_key( (string) ( $row['workflow'] ?? '' ) ) );
	}

    public function register(): void {
        add_action( Bulk_Scheduler::HOOK_TICK, [ $this, 'handle_tick' ], 10, 2 );
		add_action( 'fotogenic_ai_run_finalized', [ $this, 'sync_finalized_item' ], 30, 2 );
    }

    public function handle_tick( string $run_uuid = '', int $user_id = 0 ): void {
		if ( ! Background_Runtime_State::is_enabled() ) {
			return;
		}
        $run_uuid = trim( (string) $run_uuid );
        $user_id = (int) $user_id;
        if ( '' === $run_uuid || $user_id <= 0 ) {
            return;
        }
        $run = $this->repo->get_run_by_uuid( $run_uuid, $user_id );
        if ( empty( $run ) ) {
            return;
        }
        $run_id = (int) $run['id'];
        $status = (string) $run['status'];
        if ( ! in_array( $status, [ 'running', 'paused', 'stopping' ], true ) ) {
            return;
        }

        $owner = 'bulk-' . $run_uuid . '-' . wp_generate_uuid4();
        $item_owner = hash( 'sha256', $owner );
        if ( ! $this->repo->acquire_run_lock( $run_id, $user_id, $owner, $this->get_run_lock_seconds( $run ) ) ) {
            return;
        }
        $prev_user = get_current_user_id();

        try {
            $run = $this->repo->get_run_by_id( $run_id, $user_id );
            if ( empty( $run ) ) {
                return;
            }
            $status = (string) ( $run['status'] ?? '' );
            if ( ! in_array( $status, [ 'running', 'paused', 'stopping' ], true ) ) {
                return;
            }
            $this->repo->update_run_if_status(
                $run_id,
                $user_id,
                $status,
                [ 'last_heartbeat_at' => current_time( 'mysql', true ) ],
                $this->get_run_workflow( $run )
            );
            if ( $prev_user !== $user_id ) {
                wp_set_current_user( $user_id );
            }

            $this->start_blocked = false;
            $can_start = ( 'running' === $status );
            $now = time();
            $counts = $this->repo->count_items_by_status( $run_id, $user_id );
            $pending = (int) ( $counts['pending'] ?? 0 );
            $starting = (int) ( $counts['starting'] ?? 0 );
            $generating = (int) ( $counts['generating'] ?? 0 );
            $active_item_capacity = $this->get_available_active_item_capacity( $starting, $generating );
            $timed_out_starting = $this->fail_stuck_starting_items( $run, $now );
            if ( $timed_out_starting > 0 ) {
                $counts = $this->repo->count_items_by_status( $run_id, $user_id );
                $pending = (int) ( $counts['pending'] ?? 0 );
                $starting = (int) ( $counts['starting'] ?? 0 );
                $generating = (int) ( $counts['generating'] ?? 0 );
                $active_item_capacity = $this->get_available_active_item_capacity( $starting, $generating );
            }
            $next_delay = 30;
            $started_one = false;
            $last_start_ts = $this->get_last_start_ts( $run_uuid );
            $start_slots = $this->get_start_slots( $run_uuid, $now );

            if ( $can_start && $start_slots > 0 && $starting > 0 ) {
                $starting_items = $this->repo->list_items_by_statuses( $run_id, $user_id, [ 'starting' ], $start_slots, 'updated_at', 'ASC' );
                foreach ( $starting_items as $start_item ) {
					if ( ! $this->starting_item_belongs_to_run( $run, $start_item ) ) {
                        continue;
                    }
                    $start_item = $this->repo->claim_starting_item(
                        (int) ( $start_item['id'] ?? 0 ),
                        $user_id,
                        $this->get_run_workflow( $run ),
                        $item_owner
                    );
                    if ( ! is_array( $start_item ) ) {
                        continue;
                    }
                    $this->start_item_generation( $run, $start_item, $item_owner );
                    $started_one = true;
                    $start_slots -= 1;
                    $last_start_ts = $last_start_ts > 0 ? ( $last_start_ts + self::START_SPACING_SECONDS ) : $now;
                    if ( $this->start_blocked || $start_slots <= 0 ) {
                        break;
                    }
                }
            }
            if ( $can_start && ! $this->start_blocked && $start_slots > 0 && $pending > 0 && $active_item_capacity > 0 ) {
                $claim_limit = min( $start_slots, $active_item_capacity, $pending );
                if ( 'campaign_set' === $this->get_run_workflow( $run ) ) {
                    $claim_limit = min( 1, $claim_limit );
                }
                if ( $claim_limit > 0 ) {
                    $claimed = $this->repo->claim_pending_items( $run_id, $user_id, $claim_limit, $item_owner );
                    foreach ( $claimed as $start_item ) {
                        $this->start_item_generation( $run, $start_item, $item_owner );
                        $started_one = true;
                        $start_slots -= 1;
                        $last_start_ts = $last_start_ts > 0 ? ( $last_start_ts + self::START_SPACING_SECONDS ) : $now;
                        if ( $this->start_blocked || $start_slots <= 0 ) {
                            break;
                        }
                    }
                }
            }
            if ( $can_start && $started_one ) {
                $this->set_last_start_ts( $run_uuid, $last_start_ts );
                $this->advance_resume_ramp_if_used( $run_uuid );
                $next_delay = min( $next_delay, self::START_SPACING_SECONDS );
            }
            if ( $can_start && ( $pending > 0 || $starting > 0 ) ) {
                $spacing_delay = $this->get_start_spacing_delay( $run_uuid, $now );
                if ( $spacing_delay > 0 ) {
                    $next_delay = min( $next_delay, $spacing_delay );
                } else {
                    $next_delay = min( $next_delay, self::START_SPACING_SECONDS );
                }
            }

            $poll_items = $this->repo->list_items_by_statuses( $run_id, $user_id, [ 'generating' ], self::POLL_BATCH_LIMIT, 'last_polled_at', 'ASC' );
            foreach ( $poll_items as $poll_item ) {
                $delay = $this->get_poll_delay_seconds( $poll_item, $now );
                if ( $delay > 0 ) {
                    $next_delay = min( $next_delay, $delay );
                    continue;
                }
                $item_delay = $this->poll_item_generation( $run, $poll_item );
                if ( is_int( $item_delay ) && $item_delay > 0 ) {
                    $next_delay = min( $next_delay, $item_delay );
                }
            }

            $run = $this->repo->get_run_by_id( $run_id, $user_id );
            if ( empty( $run ) ) {
                return;
            }
            $status = (string) ( $run['status'] ?? '' );
            $workflow = $this->get_run_workflow( $run );
            $finalization_checked = $this->finalize_run_if_done( $run_id, $user_id, $status, $workflow );

            $run = $this->repo->get_run_by_id( $run_id, $user_id );
            if ( empty( $run ) ) {
                return;
            }
            $status = (string) ( $run['status'] ?? '' );
            $counts = $this->repo->count_items_by_status( $run_id, $user_id );
            $pending = (int) ( $counts['pending'] ?? 0 );
            $starting = (int) ( $counts['starting'] ?? 0 );
            $generating = (int) ( $counts['generating'] ?? 0 );
            $has_work = $generating > 0
                || ( 'running' === $status && ( $pending > 0 || $starting > 0 ) )
                || ( 'stopping' === $status && $starting > 0 )
                || ( ! $finalization_checked && in_array( $status, [ 'running', 'stopping' ], true ) );
            if ( $has_work ) {
                $this->schedule_next_tick( $run, $next_delay );
            }
        } finally {
            if ( $prev_user !== $user_id ) {
                wp_set_current_user( $prev_user );
            }
            $this->repo->release_run_lock( $run_id, $user_id, $owner );
        }
    }

    private function start_item_generation( array $run, array $item, string $item_owner ): bool {
        $run_id = (int) ( $run['id'] ?? 0 );
        $user_id = (int) ( $run['user_id'] ?? 0 );
        $item_id = (int) ( $item['id'] ?? 0 );
        $current_run = $this->repo->get_run_by_id( $run_id, $user_id );
        $stored = $this->repo->get_item_by_id( $item_id, $user_id );
        if ( ! is_array( $current_run ) || ! is_array( $stored ) ) {
            return false;
        }
        if (
            'starting' !== (string) ( $stored['status'] ?? '' )
            || ! hash_equals( $item_owner, (string) ( $stored['claim_owner'] ?? '' ) )
        ) {
            return false;
        }
        if ( $this->universal_item_dispatcher->supports( $stored ) ) {
            return $this->start_universal_item( $current_run, $stored, $item_owner );
        }

        $updated = $this->repo->update_claimed_item_if_status(
            $item_id,
            $user_id,
            $this->get_run_workflow( $current_run ),
            'starting',
            $item_owner,
            [
                'status'        => 'failed',
                'error_code'    => 'COMMAND_INVALID',
                'error_message' => __( 'The saved Bulk command is invalid.', 'fotogenic-ai' ),
                'finished_at'   => current_time( 'mysql', true ),
            ]
        );
        return $updated;
    }

}
