<?php

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

class Niwa_AI_Admin_Autonomy_Control_Plane {
    const MAX_PENDING_ACTOR_MESSAGES = 24;
    const RECENT_COMPLETED_FOUNDER_MESSAGE_TTL = 900;
    const RECENT_TEXT_DUPLICATE_FOUNDER_MESSAGE_TTL = 90;
    const MAX_RECENT_COMPLETED_FOUNDER_MESSAGES = 120;

    private $workspace;
    private $runtime_locks;

    public function __construct( $workspace = null ) {
        $this->workspace = $workspace;
        $this->runtime_locks = new Niwa_AI_Admin_Autonomy_Runtime_Locks();
    }

    public function clear_legacy_session_locks() {
        global $wpdb;

        $like_pattern = $wpdb->esc_like( 'niwa_ai_admin_turn_lock_' ) . '%';

        // phpcs:ignore WordPress.DB.DirectDatabaseQuery
        $wpdb->query(
            $wpdb->prepare(
                "DELETE FROM {$wpdb->options} WHERE option_name LIKE %s AND CAST(option_value AS UNSIGNED) < %d",
                $like_pattern,
                time()
            )
        );
    }

    public function acquire_session_lane( $session_id, $ttl = 180 ) {
        $session_id = trim( (string) $session_id );
        if ( '' === $session_id ) {
            return true;
        }

        return $this->runtime_locks->acquire( $this->session_lock_name( $session_id ), max( 30, intval( $ttl ) ) );
    }

    public function release_session_lane( $session_id ) {
        $session_id = trim( (string) $session_id );
        if ( '' === $session_id ) {
            return;
        }

        return $this->runtime_locks->release( $this->session_lock_name( $session_id ) );
    }

    public function force_release_session_lane( $session_id ) {
        $session_id = trim( (string) $session_id );
        return '' !== $session_id && $this->runtime_locks->force_release( $this->session_lock_name( $session_id ) );
    }

    public function refresh_session_lane( $session_id, $ttl = 180 ) {
        $session_id = trim( (string) $session_id );
        if ( '' === $session_id ) {
            return false;
        }

        return $this->runtime_locks->refresh( $this->session_lock_name( $session_id ), max( 30, intval( $ttl ) ) );
    }

    public function is_session_lane_locked( $session_id ) {
        $session_id = trim( (string) $session_id );
        if ( '' === $session_id ) {
            return false;
        }

        return $this->runtime_locks->is_locked( $this->session_lock_name( $session_id ) );
    }

    public function enqueue_actor_message( $session_id, $message, $meta = array() ) {
        $session_id = trim( (string) $session_id );
        if ( '' === $session_id || ! $this->has_workspace_actor_backend() ) {
            return array();
        }

        $message = $this->compact_actor_message( $message );
        if ( empty( $message ) ) {
            return array();
        }

        $actor_id = isset( $meta['actor_id'] )
            ? substr( preg_replace( '/[^a-zA-Z0-9_\-]/', '', trim( (string) $meta['actor_id'] ) ), 0, 96 )
            : '';
        $entry = array(
            'id'                    => '' !== $actor_id ? $actor_id : $this->generate_actor_message_id(),
            'enqueued_at'           => time(),
            'runtime_version'       => defined( 'NIWA_AI_VERSION' ) ? (string) NIWA_AI_VERSION : '',
            'callback_acknowledged' => ! empty( $meta['callback_acknowledged'] ),
            'message'               => $message,
        );
        $source = '';
        if ( ! empty( $meta['source'] ) ) {
            $source = sanitize_key( (string) $meta['source'] );
        } elseif ( ! empty( $message['source'] ) ) {
            $source = sanitize_key( (string) $message['source'] );
        }
        if ( '' !== $source ) {
            $entry['source'] = $source;
        }
        if ( ! empty( $meta['durable'] ) ) {
            $entry['durable'] = true;
        }
        if ( ! empty( $meta['context_payload'] ) && is_array( $meta['context_payload'] ) ) {
            $entry['context_payload'] = $meta['context_payload'];
        }
        if ( ! empty( $meta['continuation_context'] ) && is_array( $meta['continuation_context'] ) ) {
            $entry['continuation_context'] = $meta['continuation_context'];
        }
        foreach ( array(
            'continuation_depth',
            'continuation_root_actor_message_id',
            'continuation_root_update_id',
            'continuation_root_message_id',
            'continuation_parent_actor_message_id',
        ) as $continuation_meta_key ) {
            if ( array_key_exists( $continuation_meta_key, $meta ) ) {
                if ( in_array( $continuation_meta_key, array( 'continuation_depth', 'continuation_root_update_id', 'continuation_root_message_id' ), true ) ) {
                    $entry[ $continuation_meta_key ] = max( 0, intval( $meta[ $continuation_meta_key ] ) );
                } else {
                    $entry[ $continuation_meta_key ] = sanitize_text_field( (string) $meta[ $continuation_meta_key ] );
                }
            }
        }

        $lock_name = 'actor-state:' . $session_id;
        if ( ! $this->wait_for_session_lane( $lock_name, 20, 12 ) ) {
            return array();
        }

        try {
            $state = $this->normalize_actor_state(
                $session_id,
                (array) $this->workspace->read_session_actor_state( $session_id )
            );
            $cancelled_existing = false;
            $cancel_reason      = '';
            if ( $this->actor_entry_is_external_founder_message( $entry ) ) {
                $duplicate = $this->find_duplicate_actor_entry_for_message( $state, $entry );
                if ( ! empty( $duplicate['id'] ) ) {
                    $duplicate['duplicate_existing'] = true;
                    $duplicate['duplicate_actor_id'] = (string) $duplicate['id'];
                    return $duplicate;
                }

                $completed_duplicate = $this->find_recent_completed_actor_entry_for_message( $state, $entry );
                if ( ! empty( $completed_duplicate['id'] ) ) {
                    $completed_duplicate['duplicate_existing']  = true;
                    $completed_duplicate['duplicate_completed'] = true;
                    return $completed_duplicate;
                }

                if ( ! empty( $meta['supersede_existing'] ) ) {
                    $had_active_message = ! empty( $state['active_message']['id'] );
                    $state = $this->supersede_existing_actor_work_explicitly( $state );
                    if ( $had_active_message ) {
                        $cancelled_existing = true;
                        $cancel_reason      = isset( $state['cancel_reason'] ) ? (string) $state['cancel_reason'] : 'Explicit founder supersede requested.';
                    }
                }
            }
            $state['pending_messages'][] = $entry;
            $state['pending_messages']   = $this->trim_actor_pending_messages( $state['pending_messages'] );
            $state['updated_at']         = time();
            if ( ! $this->workspace->write_session_actor_state( $session_id, $state ) ) {
                return array();
            }
        } finally {
            $this->release_session_lane( $lock_name );
        }

        if ( ! empty( $cancelled_existing ) ) {
            $entry['cancelled_existing'] = true;
            $entry['cancel_reason']      = $cancel_reason;
        }

        return $entry;
    }

    public function list_sessions_with_pending_actor_messages( $limit = 10 ) {
        if ( ! $this->has_workspace_actor_backend() ) {
            return array();
        }

        $records = array();
        foreach ( (array) $this->workspace->list_session_actor_states( max( 1, min( 50, intval( $limit ) * 3 ) ) ) as $state ) {
            $state = $this->normalize_actor_state(
                isset( $state['session_id'] ) ? (string) $state['session_id'] : '',
                $state
            );
            if ( '' === $state['session_id'] ) {
                continue;
            }

            $pending_count = count( $state['pending_messages'] );
            if ( $pending_count < 1 && empty( $state['active_message'] ) ) {
                continue;
            }

            $records[] = array(
                'session_id'     => $state['session_id'],
                'pending_count'  => $pending_count,
                'has_active'     => ! empty( $state['active_message'] ),
                'active_lease_expires_at' => ! empty( $state['active_message']['lease_expires_at'] ) ? intval( $state['active_message']['lease_expires_at'] ) : 0,
                'active_last_seen_at' => ! empty( $state['active_message']['last_seen_at'] ) ? intval( $state['active_message']['last_seen_at'] ) : 0,
                'active_worker_pid' => ! empty( $state['active_message']['worker_pid'] ) ? intval( $state['active_message']['worker_pid'] ) : 0,
                'active_runtime_version' => ! empty( $state['active_message']['runtime_version'] ) ? (string) $state['active_message']['runtime_version'] : '',
                'updated_at'     => intval( $state['updated_at'] ),
                'oldest_pending' => ! empty( $state['pending_messages'][0]['enqueued_at'] ) ? intval( $state['pending_messages'][0]['enqueued_at'] ) : intval( $state['updated_at'] ),
            );
        }

        usort(
            $records,
            function ( $a, $b ) {
                return intval( $a['oldest_pending'] ) <=> intval( $b['oldest_pending'] );
            }
        );

        return array_slice( $records, 0, max( 1, min( 50, intval( $limit ) ) ) );
    }

    public function cancel_active_actor_messages( $reason = 'Active actor cancelled.', $limit = 50 ) {
        if ( ! $this->has_workspace_actor_backend() ) {
            return 0;
        }

        $cancelled = 0;
        $now       = time();
        foreach ( (array) $this->workspace->list_session_actor_states( max( 1, min( 50, intval( $limit ) ) ) ) as $raw_state ) {
            $session_id = isset( $raw_state['session_id'] ) ? trim( (string) $raw_state['session_id'] ) : '';
            if ( '' === $session_id ) {
                continue;
            }

            $lock_name = 'actor-state:' . $session_id;
            if ( ! $this->wait_for_session_lane( $lock_name, 20, 12 ) ) {
                continue;
            }

            try {
                $state = $this->normalize_actor_state(
                    $session_id,
                    (array) $this->workspace->read_session_actor_state( $session_id )
                );
                if ( empty( $state['active_message']['id'] ) ) {
                    continue;
                }

                $state['superseded_message_ids'][ (string) $state['active_message']['id'] ] = $now;
                $state['active_message'] = array();
                $state['cancelled_at']    = $now;
                $state['cancel_reason']   = sanitize_text_field( (string) $reason );
                $state['updated_at']      = $now;
                $state['superseded_message_ids'] = $this->prune_superseded_message_ids( $state['superseded_message_ids'], $now );
                $this->workspace->write_session_actor_state( $session_id, $state );
                $cancelled++;
            } finally {
                $this->release_session_lane( $lock_name );
            }
        }

        return $cancelled;
    }

    public function recover_orphaned_active_actor_messages( $reason = 'Actor runtime died while processing an active message.', $limit = 50, $max_retries = 1, $recover_live_leases = false ) {
        if ( ! $this->has_workspace_actor_backend() ) {
            return array(
                'requeued'  => 0,
                'cancelled' => 0,
                'sessions'  => array(),
            );
        }

        $result = array(
            'requeued'     => 0,
            'cancelled'    => 0,
            'skipped_live' => 0,
            'sessions'     => array(),
        );
        $now         = time();
        $max_retries = max( 0, intval( $max_retries ) );

        foreach ( (array) $this->workspace->list_session_actor_states( max( 1, min( 50, intval( $limit ) ) ) ) as $raw_state ) {
            $session_id = isset( $raw_state['session_id'] ) ? trim( (string) $raw_state['session_id'] ) : '';
            if ( '' === $session_id ) {
                continue;
            }

            $lock_name = 'actor-state:' . $session_id;
            if ( ! $this->wait_for_session_lane( $lock_name, 20, 12 ) ) {
                continue;
            }

            try {
                $state = $this->normalize_actor_state(
                    $session_id,
                    (array) $this->workspace->read_session_actor_state( $session_id )
                );
                if ( empty( $state['active_message']['id'] ) ) {
                    continue;
                }

                $active   = (array) $state['active_message'];
                $active_id = (string) $active['id'];
                $attempts = intval( isset( $active['orphan_recovery_attempts'] ) ? $active['orphan_recovery_attempts'] : 0 );
                $lease_expires_at = intval( isset( $active['lease_expires_at'] ) ? $active['lease_expires_at'] : 0 );

                if ( $this->actor_entry_has_recent_completed_identity( $state, $active ) ) {
                    $state = $this->discard_active_actor_message_as_completed_duplicate(
                        $state,
                        $active,
                        $now,
                        'Actor message recovery skipped because the same founder message already completed.'
                    );
                    $result['cancelled']++;
                    $state['updated_at'] = $now;
                    $state['superseded_message_ids'] = $this->prune_superseded_message_ids( $state['superseded_message_ids'], $now );
                    $this->workspace->write_session_actor_state( $session_id, $state );
                    $result['sessions'][] = $session_id;
                    continue;
                }

                // A live lease means the worker is alive and refreshing it. A
                // plugin runtime/version change alone must never yank an
                // in-flight turn — the old-runtime worker simply finishes it.
                // Only an explicit recover_live_leases pass (provable worker
                // death) may cross a live lease — and even then never one whose
                // own worker pid is provably alive: the death that justified
                // the pass belongs to a different process.
                if ( $lease_expires_at > $now ) {
                    $active_worker_pid = intval( isset( $active['worker_pid'] ) ? $active['worker_pid'] : 0 );
                    $worker_is_alive   = $active_worker_pid > 0
                        && class_exists( 'Niwa_AI_Admin_Autonomy_Process_Runtime' )
                        && Niwa_AI_Admin_Autonomy_Process_Runtime::background_process_is_alive( $active_worker_pid );
                    if ( empty( $recover_live_leases ) || $worker_is_alive ) {
                        $result['skipped_live']++;
                        $result['sessions'][] = $session_id;
                        continue;
                    }
                }

                if ( $this->actor_entry_is_bound_admin_ui_message( $active ) ) {
                    $state = $this->defer_bound_admin_ui_failure_recovery( $state, $active, $now, $reason );
                    if ( ! $this->workspace->write_session_actor_state( $session_id, $state ) ) {
                        continue;
                    }
                    $result['requeued']++;
                    $result['sessions'][] = $session_id;
                    continue;
                }

                unset( $active['claimed_at'], $active['lease_expires_at'], $active['cancel_generation'], $active['last_seen_at'] );
                $active['id']                       = $this->generate_actor_message_id();
                $active['enqueued_at']              = $now;
                $active['orphan_recovery_attempts'] = $attempts + 1;
                $active['orphaned_from_message_id'] = $active_id;
                $active['orphan_recovery_reason']   = sanitize_text_field( (string) $reason );
                if ( $attempts >= $max_retries ) {
                    $active['orphan_recovery_retry_limit_exceeded'] = true;
                    $active['orphan_recovery_retry_limit']          = $max_retries;
                }
                $active = $this->stamp_actor_message_current_runtime( $active );

                array_unshift( $state['pending_messages'], $active );
                $state['pending_messages'] = array_slice( $state['pending_messages'], 0, self::MAX_PENDING_ACTOR_MESSAGES );
                if ( '' !== $active_id ) {
                    $state['superseded_message_ids'][ $active_id ] = $now;
                }
                $state['cancel_generation'] = intval( isset( $state['cancel_generation'] ) ? $state['cancel_generation'] : 0 ) + 1;
                $state['active_message']   = array();
                $state['cancelled_at']     = 0;
                $state['cancel_reason']    = '';
                $result['requeued']++;
                $state['updated_at'] = $now;
                $state['superseded_message_ids'] = $this->prune_superseded_message_ids( $state['superseded_message_ids'], $now );
                $this->workspace->write_session_actor_state( $session_id, $state );
                $result['sessions'][] = $session_id;
            } finally {
                $this->release_session_lane( $lock_name );
            }
        }

        $result['sessions'] = array_values( array_unique( $result['sessions'] ) );
        return $result;
    }

    public function find_actor_message_by_scheduled_job( $session_id, $job_id, $job_uuid, $attempt_token = '' ) {
        $session_id    = trim( (string) $session_id );
        $job_id        = intval( $job_id );
        $job_uuid      = trim( (string) $job_uuid );
        $attempt_token = trim( (string) $attempt_token );
        if ( '' === $session_id || $job_id <= 0 || '' === $job_uuid || ! $this->has_workspace_actor_backend() ) {
            return array(
                'found' => false,
            );
        }

        $lock_name = 'actor-state:' . $session_id;
        if ( ! $this->wait_for_session_lane( $lock_name, 20, 12 ) ) {
            return array(
                'found' => false,
                'error' => 'actor_state_lock_unavailable',
            );
        }

        try {
            $state = $this->normalize_actor_state(
                $session_id,
                (array) $this->workspace->read_session_actor_state( $session_id )
            );
            $candidates = array();
            if ( ! empty( $state['active_message'] ) && is_array( $state['active_message'] ) ) {
                $candidates[] = array(
                    'state' => 'active',
                    'entry' => $state['active_message'],
                );
            }
            foreach ( $state['pending_messages'] as $pending ) {
                if ( is_array( $pending ) ) {
                    $candidates[] = array(
                        'state' => 'pending',
                        'entry' => $pending,
                    );
                }
            }

            foreach ( $candidates as $candidate ) {
                $entry = isset( $candidate['entry'] ) && is_array( $candidate['entry'] ) ? $candidate['entry'] : array();
                $ref   = $this->actor_entry_scheduled_job_reference( $entry );
                if ( empty( $ref['id'] ) || intval( $ref['id'] ) !== $job_id ) {
                    continue;
                }
                if ( empty( $ref['job_uuid'] ) || ! hash_equals( $job_uuid, (string) $ref['job_uuid'] ) ) {
                    continue;
                }
                if ( '' !== $attempt_token && ! $this->actor_entry_scheduled_attempt_matches( $entry, $attempt_token ) ) {
                    continue;
                }

                return array(
                    'found'            => true,
                    'state'            => isset( $candidate['state'] ) ? (string) $candidate['state'] : '',
                    'id'               => isset( $entry['id'] ) ? (string) $entry['id'] : '',
                    'lease_expires_at' => intval( isset( $entry['lease_expires_at'] ) ? $entry['lease_expires_at'] : 0 ),
                    'claimed_at'       => intval( isset( $entry['claimed_at'] ) ? $entry['claimed_at'] : 0 ),
                    'runtime_current'  => $this->actor_message_runtime_is_current( $entry ),
                    'entry'            => $entry,
                );
            }
        } finally {
            $this->release_session_lane( $lock_name );
        }

        return array(
            'found' => false,
        );
    }

    public function checkpoint_active_actor_message( $session_id, $message_id, $continuation, $reason = 'soft_deadline' ) {
        $session_id = trim( (string) $session_id );
        $message_id = trim( (string) $message_id );
        if ( '' === $session_id || '' === $message_id || ! $this->has_workspace_actor_backend() ) {
            return array(
                'success' => false,
                'error'   => 'checkpoint_invalid_actor',
            );
        }

        $continuation = is_array( $continuation ) ? $continuation : array();
        $chat_id      = isset( $continuation['chat_id'] ) ? trim( (string) $continuation['chat_id'] ) : '';
        if ( '' === $chat_id ) {
            return array(
                'success' => false,
                'error'   => 'checkpoint_missing_chat_id',
            );
        }

        $lock_name = 'actor-state:' . $session_id;
        if ( ! $this->wait_for_session_lane( $lock_name, 20, 12 ) ) {
            return array(
                'success' => false,
                'error'   => 'checkpoint_actor_state_locked',
            );
        }

        try {
            $state = $this->normalize_actor_state(
                $session_id,
                (array) $this->workspace->read_session_actor_state( $session_id )
            );

            if ( empty( $state['active_message']['id'] ) || (string) $state['active_message']['id'] !== $message_id ) {
                return array(
                    'success' => false,
                    'error'   => 'checkpoint_active_message_mismatch',
                );
            }

            $now    = time();
            $active = (array) $state['active_message'];
            $continuation_meta = $this->build_continuation_actor_metadata( $active, $continuation, $message_id );
            $entry = array(
                'id'                         => $this->generate_actor_message_id(),
                'enqueued_at'                => $now,
                'runtime_version'            => defined( 'NIWA_AI_VERSION' ) ? (string) NIWA_AI_VERSION : '',
                'callback_acknowledged'      => true,
                'durable'                    => true,
                'source'                     => 'actor_continuation',
                'checkpointed_from_message_id' => $message_id,
                'checkpoint_reason'          => sanitize_text_field( (string) $reason ),
                'continuation_depth'         => $continuation_meta['depth'],
                'continuation_root_actor_message_id' => $continuation_meta['root_actor_message_id'],
                'continuation_root_update_id' => $continuation_meta['root_update_id'],
                'continuation_root_message_id' => $continuation_meta['root_message_id'],
                'continuation_parent_actor_message_id' => $message_id,
                'message'                    => array(
                    'channel'    => isset( $continuation['channel'] )
                        ? sanitize_key( (string) $continuation['channel'] )
                        : ( isset( $message['channel'] ) ? sanitize_key( (string) $message['channel'] ) : 'telegram' ),
                    'chat_id'    => $chat_id,
                    'message_id' => $continuation_meta['message_id'],
                    'update_id'  => $continuation_meta['update_id'],
                    'text'       => '[continuation]',
                    'user_role'  => isset( $continuation['user_role'] ) ? (string) $continuation['user_role'] : 'admin',
                ),
            );

            if ( ! empty( $continuation['context_payload'] ) && is_array( $continuation['context_payload'] ) ) {
                $entry['context_payload'] = $continuation['context_payload'];
            }
            if ( ! empty( $continuation['context_snapshot'] ) && is_array( $continuation['context_snapshot'] ) ) {
                $entry['continuation_context'] = $continuation['context_snapshot'];
            }

            $entry = $this->stamp_actor_message_current_runtime( $entry );
            $state['pending_messages'] = $this->insert_actor_continuation_after_scheduled_work( $state['pending_messages'], $entry );
            $state['pending_messages'] = $this->trim_actor_pending_messages( $state['pending_messages'] );
            $state['active_message']   = array();
            $state['cancelled_at']     = 0;
            $state['cancel_reason']    = '';
            $state['updated_at']       = $now;
            if ( ! $this->workspace->write_session_actor_state( $session_id, $state ) ) {
                return array(
                    'success' => false,
                    'error'   => 'checkpoint_actor_state_write_failed',
                );
            }

            return array(
                'success' => true,
                'entry'   => $entry,
            );
        } finally {
            $this->release_session_lane( $lock_name );
        }
    }

    private function insert_actor_continuation_after_scheduled_work( $messages, $entry ) {
        $messages = array_values( (array) $messages );
        $entry    = is_array( $entry ) ? $entry : array();

        $insert_at = 0;
        foreach ( $messages as $index => $pending ) {
            if ( $this->actor_entry_is_scheduled_self_wake_work( $pending ) ) {
                $insert_at = $index + 1;
            }
        }

        array_splice( $messages, $insert_at, 0, array( $entry ) );
        return $messages;
    }

    private function build_continuation_actor_metadata( $active, $continuation, $parent_message_id ) {
        $active       = is_array( $active ) ? $active : array();
        $continuation = is_array( $continuation ) ? $continuation : array();
        $message      = isset( $active['message'] ) && is_array( $active['message'] ) ? $active['message'] : array();

        $update_id  = intval( isset( $continuation['update_id'] ) ? $continuation['update_id'] : ( isset( $message['update_id'] ) ? $message['update_id'] : 0 ) );
        $message_id = intval( isset( $continuation['message_id'] ) ? $continuation['message_id'] : ( isset( $message['message_id'] ) ? $message['message_id'] : 0 ) );
        $root_actor = isset( $active['continuation_root_actor_message_id'] ) && '' !== trim( (string) $active['continuation_root_actor_message_id'] )
            ? trim( (string) $active['continuation_root_actor_message_id'] )
            : trim( (string) $parent_message_id );

        return array(
            'depth'                 => intval( isset( $active['continuation_depth'] ) ? $active['continuation_depth'] : 0 ) + 1,
            'root_actor_message_id' => $root_actor,
            'root_update_id'        => intval( isset( $active['continuation_root_update_id'] ) ? $active['continuation_root_update_id'] : $update_id ),
            'root_message_id'       => intval( isset( $active['continuation_root_message_id'] ) ? $active['continuation_root_message_id'] : $message_id ),
            'update_id'             => $update_id,
            'message_id'            => $message_id,
        );
    }

    public function claim_next_actor_message( $session_id, $lease_ttl = 600 ) {
        $session_id = trim( (string) $session_id );
        if ( '' === $session_id || ! $this->has_workspace_actor_backend() ) {
            return array();
        }

        $lock_name = 'actor-state:' . $session_id;
        if ( ! $this->wait_for_session_lane( $lock_name, 20, 12 ) ) {
            return array();
        }

        try {
            $state = $this->normalize_actor_state(
                $session_id,
                (array) $this->workspace->read_session_actor_state( $session_id )
            );
            $now   = time();

            if ( ! empty( $state['active_message'] ) ) {
                $active           = (array) $state['active_message'];
                $active_id        = isset( $active['id'] ) ? (string) $active['id'] : '';
                $lease_expires_at = intval( isset( $active['lease_expires_at'] ) ? $active['lease_expires_at'] : 0 );
                if ( '' !== $active_id && $this->actor_message_is_superseded( $state, $active_id ) ) {
                    $state['superseded_message_ids'][ $active_id ] = $now;
                    $state['active_message'] = array();
                    $state['cancelled_at']    = $now;
                    $state['cancel_reason']   = 'Actor message superseded.';
                    $state['updated_at']     = $now;
                    $state['superseded_message_ids'] = $this->prune_superseded_message_ids( $state['superseded_message_ids'], $now );
                    $this->workspace->write_session_actor_state( $session_id, $state );
                    $active = array();
                } elseif ( ! $this->actor_message_runtime_is_current( $active ) && $lease_expires_at <= $now ) {
                    // Only reclaim old-runtime work once its lease has lapsed.
                    // While the lease is live the old-runtime worker is still
                    // processing it and must be left to finish its turn.
                    $state = $this->requeue_expired_active_actor_message(
                        $state,
                        $active,
                        $now,
                        'Actor message lease expired after a plugin runtime change.'
                    );
                    $this->workspace->write_session_actor_state( $session_id, $state );
                }

                if ( ! empty( $active ) && ! empty( $state['active_message'] ) && $this->actor_entry_has_recent_completed_identity( $state, $active ) ) {
                    $state = $this->discard_active_actor_message_as_completed_duplicate(
                        $state,
                        $active,
                        $now,
                        'Actor message abandoned because the same founder message already completed.'
                    );
                    $this->workspace->write_session_actor_state( $session_id, $state );
                }
            }

            if ( ! empty( $state['active_message'] ) ) {
                $active           = (array) $state['active_message'];
                $active_id        = isset( $active['id'] ) ? (string) $active['id'] : '';
                $lease_expires_at = intval( isset( $active['lease_expires_at'] ) ? $active['lease_expires_at'] : 0 );
                if ( $lease_expires_at > $now ) {
                    return array();
                }

                $state = $this->requeue_expired_active_actor_message(
                    $state,
                    $active,
                    $now,
                    'Actor message lease expired before completion.'
                );
                $this->workspace->write_session_actor_state( $session_id, $state );
            }

			if ( empty( $state['pending_messages'] ) ) {
				$state['updated_at'] = $now;
				$this->workspace->write_session_actor_state( $session_id, $state );
				return array();
			}

			$state['pending_messages'] = $this->prioritize_external_founder_actor_messages( $state['pending_messages'] );

			$entry          = array();
			$scan_remaining = count( $state['pending_messages'] );
			while ( $scan_remaining > 0 && ! empty( $state['pending_messages'] ) ) {
				$candidate = array_shift( $state['pending_messages'] );
				$scan_remaining--;
				if ( ! is_array( $candidate ) ) {
					continue;
				}
				if ( ! empty( $candidate['not_before'] ) && intval( $candidate['not_before'] ) > $now ) {
					$state['pending_messages'][] = $candidate;
					continue;
				}
				$entry = $candidate;
				if ( ! empty( $entry['id'] ) && ! $this->actor_message_runtime_is_current( $entry ) ) {
					$state['superseded_message_ids'][ (string) $entry['id'] ] = $now;
					$state['cancelled_at']  = $now;
                    $state['cancel_reason'] = 'Actor message abandoned because it belongs to an older plugin runtime.';
                    $entry = array();
                } elseif ( ! empty( $entry ) && $this->actor_entry_has_recent_completed_identity( $state, $entry ) ) {
                    if ( ! empty( $entry['id'] ) ) {
                        $state['superseded_message_ids'][ (string) $entry['id'] ] = $now;
                    }
                    $state = $this->remove_pending_actor_entries_for_message_identity(
                        $state,
                        $entry,
                        $now
                    );
                    $state['cancelled_at']  = $now;
                    $state['cancel_reason'] = 'Actor message skipped because the same founder message already completed.';
					$entry = array();
				}
				if ( ! empty( $entry ) && ( empty( $entry['id'] ) || ! $this->actor_message_is_superseded( $state, (string) $entry['id'] ) ) ) {
					break;
				}
				$entry = array();
			}

            if ( empty( $entry ) || ( ! empty( $entry['id'] ) && $this->actor_message_is_superseded( $state, (string) $entry['id'] ) ) ) {
                $state['updated_at'] = $now;
                $state['superseded_message_ids'] = $this->prune_superseded_message_ids( $state['superseded_message_ids'], $now );
                $this->workspace->write_session_actor_state( $session_id, $state );
                return array();
            }

            $entry['claimed_at']       = $now;
            $entry['lease_expires_at'] = $now + max( 60, intval( $lease_ttl ) );
            $entry['cancel_generation'] = intval( isset( $state['cancel_generation'] ) ? $state['cancel_generation'] : 0 );
            $entry['worker_pid']       = function_exists( 'getmypid' ) ? max( 0, intval( getmypid() ) ) : 0;
            unset( $entry['not_before'] );
            $entry = $this->stamp_actor_message_current_runtime( $entry );
            $state = $this->remove_pending_actor_entries_for_message_identity(
                $state,
                $entry,
                $now
            );
            $state['active_message']   = $entry;
            $state['updated_at']       = $now;
            if ( ! $this->workspace->write_session_actor_state( $session_id, $state ) ) {
                return array();
            }

            return $entry;
        } finally {
            $this->release_session_lane( $lock_name );
        }
    }

    public function record_actor_message_completion( $session_id, $entry ) {
        $session_id = trim( (string) $session_id );
        $entry      = is_array( $entry ) ? $entry : array();
        if ( '' === $session_id || empty( $entry ) || ! $this->has_workspace_actor_backend() ) {
            return false;
        }

        $identity = $this->actor_entry_founder_message_identity_key( $entry );
        if ( '' === $identity ) {
            return false;
        }

        $lock_name = 'actor-state:' . $session_id;
        if ( ! $this->wait_for_session_lane( $lock_name, 20, 40 ) ) {
            return false;
        }

        try {
            $state = $this->normalize_actor_state(
                $session_id,
                (array) $this->workspace->read_session_actor_state( $session_id )
            );
            $now      = time();
            $entry_id = isset( $entry['id'] ) ? (string) $entry['id'] : '';

            $state = $this->remember_recent_completed_actor_message( $state, $entry );
            $state = $this->remove_pending_actor_entries_for_message_identity( $state, $entry, $now );

            if ( '' !== $entry_id && ! empty( $state['active_message']['id'] ) && (string) $state['active_message']['id'] === $entry_id ) {
                $state['active_message']    = array();
                $state['last_completed_at'] = $now;
            }

            $state['updated_at'] = $now;
            return ! empty( $this->workspace->write_session_actor_state( $session_id, $state ) );
        } finally {
            $this->release_session_lane( $lock_name );
        }
    }

    public function complete_actor_message( $session_id, $message_id ) {
        $session_id = trim( (string) $session_id );
        $message_id = trim( (string) $message_id );
        if ( '' === $session_id || '' === $message_id || ! $this->has_workspace_actor_backend() ) {
            return false;
        }

        $lock_name = 'actor-state:' . $session_id;
        if ( ! $this->wait_for_session_lane( $lock_name, 20, 12 ) ) {
            return false;
        }

        try {
            $state = $this->normalize_actor_state(
                $session_id,
                (array) $this->workspace->read_session_actor_state( $session_id )
            );
            if ( ! empty( $state['active_message']['id'] ) && (string) $state['active_message']['id'] === $message_id ) {
                $completed_entry = (array) $state['active_message'];
                $completed_at    = time();
                $state = $this->remember_recent_completed_actor_message( $state, $completed_entry );
                $state = $this->remove_pending_actor_entries_for_message_identity(
                    $state,
                    $completed_entry,
                    $completed_at
                );
                $state['active_message']    = array();
                $state['last_completed_at'] = $completed_at;
                $state['updated_at']        = $completed_at;
                return ! empty( $this->workspace->write_session_actor_state( $session_id, $state ) );
            }
        } finally {
            $this->release_session_lane( $lock_name );
        }

        return true;
    }

    public function requeue_active_actor_message( $session_id, $message_id, $reason = 'Actor runtime failed while processing an active message.' ) {
        $session_id = trim( (string) $session_id );
        $message_id = trim( (string) $message_id );
        if ( '' === $session_id || '' === $message_id || ! $this->has_workspace_actor_backend() ) {
            return array(
                'success'  => false,
                'requeued' => 0,
            );
        }

        $lock_name = 'actor-state:' . $session_id;
        if ( ! $this->wait_for_session_lane( $lock_name, 20, 12 ) ) {
            return array(
                'success'  => false,
                'requeued' => 0,
            );
        }

        try {
            $state = $this->normalize_actor_state(
                $session_id,
                (array) $this->workspace->read_session_actor_state( $session_id )
            );
            if ( empty( $state['active_message']['id'] ) || (string) $state['active_message']['id'] !== $message_id ) {
                return array(
                    'success'  => false,
                    'requeued' => 0,
                );
            }

            $state = $this->requeue_expired_active_actor_message(
                $state,
                (array) $state['active_message'],
                time(),
                $reason
            );
            $this->workspace->write_session_actor_state( $session_id, $state );

            return array(
                'success'  => true,
                'requeued' => 1,
            );
        } finally {
            $this->release_session_lane( $lock_name );
        }
    }

    public function defer_active_actor_message( $session_id, $message_id, $delay_seconds = 5, $updates = array() ) {
        $session_id = trim( (string) $session_id );
        $message_id = trim( (string) $message_id );
        if ( '' === $session_id || '' === $message_id || ! $this->has_workspace_actor_backend() ) {
            return false;
        }

        $lock_name = 'actor-state:' . $session_id;
        if ( ! $this->wait_for_session_lane( $lock_name, 20, 12 ) ) {
            return false;
        }

        try {
            $state = $this->normalize_actor_state(
                $session_id,
                (array) $this->workspace->read_session_actor_state( $session_id )
            );
            $active = isset( $state['active_message'] ) && is_array( $state['active_message'] ) ? $state['active_message'] : array();
            if ( empty( $active['id'] ) || ! hash_equals( $message_id, (string) $active['id'] ) ) {
                return false;
            }

            $now = time();
            unset( $active['claimed_at'], $active['lease_expires_at'], $active['cancel_generation'], $active['last_seen_at'], $active['worker_pid'] );
            $updates = is_array( $updates ) ? $updates : array();
            if ( ! empty( $updates['admin_ui_failure_pending'] ) ) {
                $active['admin_ui_failure_pending']  = true;
                $active['admin_ui_failure_attempts'] = max( 1, intval( $updates['admin_ui_failure_attempts'] ?? 1 ) );
                $active['admin_ui_failure_reason']   = sanitize_text_field( (string) ( $updates['admin_ui_failure_reason'] ?? '' ) );
            }
            $active['enqueued_at'] = $now;
            $active['not_before']  = $now + max( 1, min( 300, intval( $delay_seconds ) ) );
            array_unshift( $state['pending_messages'], $active );
            $state['pending_messages'] = $this->trim_actor_pending_messages( $state['pending_messages'] );
            $state['active_message']   = array();
            $state['updated_at']       = time();

            return ! empty( $this->workspace->write_session_actor_state( $session_id, $state ) );
        } finally {
            $this->release_session_lane( $lock_name );
        }
    }

    public function cancel_actor_session( $session_id, $reason = 'Cancelled by founder.' ) {
        $session_id = trim( (string) $session_id );
        if ( '' === $session_id || ! $this->has_workspace_actor_backend() ) {
            return array(
                'success' => false,
                'error'   => 'Actor backend is not available.',
            );
        }

        $lock_name = 'actor-state:' . $session_id;
        if ( ! $this->wait_for_session_lane( $lock_name, 20, 12 ) ) {
            return array(
                'success' => false,
                'error'   => 'Actor state is locked.',
            );
        }

        $pending_cancelled = 0;
        $active_cancelled  = 0;
        try {
            $state = $this->normalize_actor_state(
                $session_id,
                (array) $this->workspace->read_session_actor_state( $session_id )
            );
            $now = time();

            foreach ( $state['pending_messages'] as $entry ) {
                if ( ! empty( $entry['id'] ) ) {
                    $state['superseded_message_ids'][ (string) $entry['id'] ] = $now;
                    $pending_cancelled++;
                }
            }
            $state['pending_messages'] = array();

            if ( ! empty( $state['active_message']['id'] ) ) {
                $state['superseded_message_ids'][ (string) $state['active_message']['id'] ] = $now;
                $state['active_message'] = array();
                $active_cancelled = 1;
            }

            $state['cancelled_at']    = $now;
            $state['cancel_reason']   = (string) $reason;
            $state['cancel_generation'] = intval( isset( $state['cancel_generation'] ) ? $state['cancel_generation'] : 0 ) + 1;
            $state['updated_at']      = $now;
            $state['superseded_message_ids'] = $this->prune_superseded_message_ids( $state['superseded_message_ids'], $now );
            $this->workspace->write_session_actor_state( $session_id, $state );
        } finally {
            $this->release_session_lane( $lock_name );
        }

        return array(
            'success'           => true,
            'pending_cancelled' => $pending_cancelled,
            'active_cancelled'  => $active_cancelled,
        );
    }

    public function is_actor_message_superseded( $session_id, $message_id ) {
        $session_id = trim( (string) $session_id );
        $message_id = trim( (string) $message_id );
        if ( '' === $session_id || '' === $message_id || ! $this->has_workspace_actor_backend() ) {
            return false;
        }

        $state = $this->normalize_actor_state(
            $session_id,
            (array) $this->workspace->read_session_actor_state( $session_id )
        );

        return $this->actor_message_is_superseded( $state, $message_id );
    }

    public function is_actor_message_cancelled( $session_id, $message_id, $cancel_generation = 0 ) {
        $session_id = trim( (string) $session_id );
        $message_id = trim( (string) $message_id );
        if ( '' === $session_id || '' === $message_id || ! $this->has_workspace_actor_backend() ) {
            return false;
        }

        $state = $this->normalize_actor_state(
            $session_id,
            (array) $this->workspace->read_session_actor_state( $session_id )
        );

        if ( $this->actor_message_is_superseded( $state, $message_id ) ) {
            return true;
        }

        return false;
    }

    public function refresh_actor_message_lease( $session_id, $message_id, $lease_ttl = 90 ) {
        $session_id = trim( (string) $session_id );
        $message_id = trim( (string) $message_id );
        if ( '' === $session_id || '' === $message_id || ! $this->has_workspace_actor_backend() ) {
            return false;
        }

        $lock_name = 'actor-state:' . $session_id;
        if ( ! $this->wait_for_session_lane( $lock_name, 20, 12 ) ) {
            return false;
        }

        try {
            $state = $this->normalize_actor_state(
                $session_id,
                (array) $this->workspace->read_session_actor_state( $session_id )
            );
            if ( empty( $state['active_message']['id'] ) || (string) $state['active_message']['id'] !== $message_id ) {
                return false;
            }

            $now = time();
            $state['active_message']['last_seen_at']    = $now;
            $state['active_message']['lease_expires_at'] = $now + max( 60, intval( $lease_ttl ) );
            $state['active_message']['worker_pid']       = function_exists( 'getmypid' ) ? max( 0, intval( getmypid() ) ) : 0;
            $state['updated_at']                        = $now;
            $this->workspace->write_session_actor_state( $session_id, $state );
        } finally {
            $this->release_session_lane( $lock_name );
        }

        return true;
    }

    public function has_pending_actor_messages() {
        return ! empty( $this->list_sessions_with_pending_actor_messages( 1 ) );
    }

    private function has_workspace_actor_backend() {
        return $this->workspace
            && method_exists( $this->workspace, 'read_session_actor_state' )
            && method_exists( $this->workspace, 'write_session_actor_state' )
            && method_exists( $this->workspace, 'list_session_actor_states' );
    }

    private function normalize_actor_state( $session_id, $state ) {
        $state = is_array( $state ) ? $state : array();
        return array(
            'session_id'        => trim( (string) $session_id ),
            'updated_at'        => intval( isset( $state['updated_at'] ) ? $state['updated_at'] : 0 ),
            'pending_messages'  => isset( $state['pending_messages'] ) && is_array( $state['pending_messages'] ) ? array_values( $state['pending_messages'] ) : array(),
            'active_message'    => isset( $state['active_message'] ) && is_array( $state['active_message'] ) ? $state['active_message'] : array(),
            'last_completed_at' => intval( isset( $state['last_completed_at'] ) ? $state['last_completed_at'] : 0 ),
            'superseded_message_ids' => isset( $state['superseded_message_ids'] ) && is_array( $state['superseded_message_ids'] ) ? $this->normalize_superseded_message_ids( $state['superseded_message_ids'] ) : array(),
            'recent_completed_message_identities' => isset( $state['recent_completed_message_identities'] ) && is_array( $state['recent_completed_message_identities'] ) ? $this->normalize_recent_completed_message_identities( $state['recent_completed_message_identities'] ) : array(),
            'cancelled_at'      => intval( isset( $state['cancelled_at'] ) ? $state['cancelled_at'] : 0 ),
            'cancel_reason'     => isset( $state['cancel_reason'] ) ? (string) $state['cancel_reason'] : '',
            'cancel_generation' => intval( isset( $state['cancel_generation'] ) ? $state['cancel_generation'] : 0 ),
        );
    }

    private function find_duplicate_actor_entry_for_message( $state, $entry ) {
        $state = $this->normalize_actor_state(
            isset( $state['session_id'] ) ? (string) $state['session_id'] : '',
            $state
        );

        $candidates = array();
        if ( ! empty( $state['active_message'] ) && is_array( $state['active_message'] ) ) {
            $candidates[] = $state['active_message'];
        }
        foreach ( $state['pending_messages'] as $pending ) {
            if ( is_array( $pending ) ) {
                $candidates[] = $pending;
            }
        }

        foreach ( $candidates as $candidate ) {
            if ( $this->actor_entries_have_same_founder_message_identity( $candidate, $entry ) || $this->actor_entries_have_same_recent_founder_text_fingerprint( $candidate, $entry, time() ) ) {
                return $candidate;
            }
        }

        return array();
    }

    private function find_recent_completed_actor_entry_for_message( $state, $entry ) {
        $identity    = $this->actor_entry_founder_message_identity_key( $entry );
        $fingerprint = $this->actor_entry_founder_message_text_fingerprint_key( $entry );
        if ( '' === $identity && '' === $fingerprint ) {
            return array();
        }

        $state = $this->normalize_actor_state(
            isset( $state['session_id'] ) ? (string) $state['session_id'] : '',
            $state
        );

        $completed_identity = $identity;
        $completed          = '' !== $identity && isset( $state['recent_completed_message_identities'][ $identity ] ) && is_array( $state['recent_completed_message_identities'][ $identity ] )
            ? $state['recent_completed_message_identities'][ $identity ]
            : array();
        if ( empty( $completed ) && '' !== $fingerprint && isset( $state['recent_completed_message_identities'][ $fingerprint ] ) && is_array( $state['recent_completed_message_identities'][ $fingerprint ] ) ) {
            $candidate_completed_at = intval( isset( $state['recent_completed_message_identities'][ $fingerprint ]['completed_at'] ) ? $state['recent_completed_message_identities'][ $fingerprint ]['completed_at'] : 0 );
            if ( $candidate_completed_at >= time() - self::RECENT_TEXT_DUPLICATE_FOUNDER_MESSAGE_TTL ) {
                $completed_identity = $fingerprint;
                $completed          = $state['recent_completed_message_identities'][ $fingerprint ];
            }
        }

        if ( empty( $completed ) ) {
            return array();
        }

        return array(
            'id'                    => isset( $completed['actor_id'] ) ? (string) $completed['actor_id'] : $identity,
            'completed_at'          => intval( isset( $completed['completed_at'] ) ? $completed['completed_at'] : 0 ),
            'duplicate_identity'    => $completed_identity,
            'message'               => isset( $entry['message'] ) && is_array( $entry['message'] ) ? $entry['message'] : array(),
        );
    }

    private function actor_entry_has_recent_completed_identity( $state, $entry ) {
        return ! empty( $this->find_recent_completed_actor_entry_for_message( $state, $entry ) );
    }

    private function remove_pending_actor_entries_for_message_identity( $state, $entry, $now ) {
        $state = $this->normalize_actor_state(
            isset( $state['session_id'] ) ? (string) $state['session_id'] : '',
            $state
        );

        $kept = array();
        foreach ( $state['pending_messages'] as $pending ) {
            if ( is_array( $pending ) && ( $this->actor_entries_have_same_founder_message_identity( $pending, $entry ) || $this->actor_entries_have_same_recent_founder_text_fingerprint( $pending, $entry, intval( $now ) ) ) ) {
                if ( ! empty( $pending['id'] ) ) {
                    $state['superseded_message_ids'][ (string) $pending['id'] ] = intval( $now );
                }
                continue;
            }
            $kept[] = $pending;
        }

        $state['pending_messages'] = $kept;
        $state['superseded_message_ids'] = $this->prune_superseded_message_ids(
            $state['superseded_message_ids'],
            intval( $now )
        );

        return $state;
    }

    private function discard_active_actor_message_as_completed_duplicate( $state, $active, $now, $reason ) {
        $state = $this->normalize_actor_state(
            isset( $state['session_id'] ) ? (string) $state['session_id'] : '',
            $state
        );

        if ( ! empty( $active['id'] ) ) {
            $state['superseded_message_ids'][ (string) $active['id'] ] = intval( $now );
        }
        $state = $this->remove_pending_actor_entries_for_message_identity( $state, $active, $now );
        $state['active_message']    = array();
        $state['cancelled_at']      = intval( $now );
        $state['cancel_reason']     = sanitize_text_field( (string) $reason );
        $state['cancel_generation'] = intval( isset( $state['cancel_generation'] ) ? $state['cancel_generation'] : 0 ) + 1;
        $state['updated_at']        = intval( $now );

        return $state;
    }

    private function actor_entries_have_same_founder_message_identity( $first, $second ) {
        $first_key  = $this->actor_entry_founder_message_identity_key( $first );
        $second_key = $this->actor_entry_founder_message_identity_key( $second );

        return '' !== $first_key && $first_key === $second_key;
    }

    private function actor_entries_have_same_recent_founder_text_fingerprint( $first, $second, $now ) {
        $first_key  = $this->actor_entry_founder_message_text_fingerprint_key( $first );
        $second_key = $this->actor_entry_founder_message_text_fingerprint_key( $second );
        if ( '' === $first_key || $first_key !== $second_key ) {
            return false;
        }

        $first_seen  = $this->actor_entry_message_seen_at( $first );
        $second_seen = $this->actor_entry_message_seen_at( $second );
        if ( $first_seen <= 0 || $second_seen <= 0 ) {
            return true;
        }

        return abs( $first_seen - $second_seen ) <= self::RECENT_TEXT_DUPLICATE_FOUNDER_MESSAGE_TTL
            || max( $first_seen, $second_seen ) >= intval( $now ) - self::RECENT_TEXT_DUPLICATE_FOUNDER_MESSAGE_TTL;
    }

    private function actor_entry_founder_message_identity_key( $entry ) {
        $entry   = is_array( $entry ) ? $entry : array();
        $message = isset( $entry['message'] ) && is_array( $entry['message'] ) ? $entry['message'] : array();
        $channel = isset( $message['channel'] ) ? sanitize_key( (string) $message['channel'] ) : '';
        $source  = isset( $entry['source'] ) ? sanitize_key( (string) $entry['source'] ) : '';
        if ( 'actor_continuation' === $source ) {
            return '';
        }

        $update_id = isset( $message['update_id'] ) ? intval( $message['update_id'] ) : 0;
        if ( $update_id > 0 ) {
            return ( '' !== $channel ? $channel : 'unknown' ) . ':update:' . $update_id;
        }

        $chat_id    = isset( $message['chat_id'] ) ? trim( (string) $message['chat_id'] ) : '';
        $message_id = isset( $message['message_id'] ) ? intval( $message['message_id'] ) : 0;
        if ( '' !== $chat_id && $message_id > 0 ) {
            return ( '' !== $channel ? $channel : 'unknown' ) . ':message:' . $chat_id . ':' . $message_id;
        }

        $raw_message_id = isset( $message['raw_message_id'] ) ? trim( (string) $message['raw_message_id'] ) : '';
        if ( '' !== $chat_id && '' !== $raw_message_id ) {
            return ( '' !== $channel ? $channel : 'unknown' ) . ':raw:' . $chat_id . ':' . md5( $raw_message_id );
        }

        return '';
    }

    private function actor_entry_founder_message_text_fingerprint_key( $entry ) {
        if ( ! $this->actor_entry_is_external_founder_message( $entry ) ) {
            return '';
        }

        $entry   = is_array( $entry ) ? $entry : array();
        $message = isset( $entry['message'] ) && is_array( $entry['message'] ) ? $entry['message'] : array();
        $text    = '';
        if ( isset( $message['text'] ) ) {
            $text = trim( (string) $message['text'] );
        }
        if ( '' === $text && isset( $message['caption'] ) ) {
            $text = trim( (string) $message['caption'] );
        }
        if ( '' === $text ) {
            return '';
        }

        foreach ( array( 'media_refs', 'reply_media_refs', 'album_message_ids', 'fragment_message_ids', 'callback_query_id', 'callback_data' ) as $field ) {
            if ( ! empty( $message[ $field ] ) ) {
                return '';
            }
        }

        $channel = isset( $message['channel'] ) ? sanitize_key( (string) $message['channel'] ) : 'unknown';
        if ( 'admin_ui' === $channel ) {
            return '';
        }
        $chat_id = isset( $message['chat_id'] ) ? trim( (string) $message['chat_id'] ) : '';
        if ( '' === $chat_id ) {
            return '';
        }

        $normalized = strtolower( $text );
        if ( function_exists( 'remove_accents' ) ) {
            $normalized = remove_accents( $normalized );
        }
        $normalized = preg_replace( '/\s+/', ' ', trim( (string) $normalized ) );
        if ( '' === $normalized ) {
            return '';
        }

        return ( '' !== $channel ? $channel : 'unknown' ) . ':text-fingerprint:' . $chat_id . ':' . md5( $normalized );
    }

    private function actor_entry_message_seen_at( $entry ) {
        $entry   = is_array( $entry ) ? $entry : array();
        $message = isset( $entry['message'] ) && is_array( $entry['message'] ) ? $entry['message'] : array();
        foreach ( array(
            isset( $message['date'] ) ? $message['date'] : null,
            isset( $entry['enqueued_at'] ) ? $entry['enqueued_at'] : null,
            isset( $entry['claimed_at'] ) ? $entry['claimed_at'] : null,
        ) as $candidate ) {
            $timestamp = intval( $candidate );
            if ( $timestamp > 0 ) {
                return $timestamp;
            }
        }

        return 0;
    }

    private function actor_entry_scheduled_job_reference( $entry ) {
        $entry      = is_array( $entry ) ? $entry : array();
        $candidates = array();
        if ( ! empty( $entry['message']['scheduled_job'] ) && is_array( $entry['message']['scheduled_job'] ) ) {
            $candidates[] = $entry['message']['scheduled_job'];
        }
        if ( ! empty( $entry['context_payload']['scheduled_job'] ) && is_array( $entry['context_payload']['scheduled_job'] ) ) {
            $candidates[] = $entry['context_payload']['scheduled_job'];
        }
        if ( ! empty( $entry['continuation_context']['scheduled_job'] ) && is_array( $entry['continuation_context']['scheduled_job'] ) ) {
            $candidates[] = $entry['continuation_context']['scheduled_job'];
        }

        foreach ( $candidates as $candidate ) {
            $id       = isset( $candidate['id'] ) ? intval( $candidate['id'] ) : 0;
            $job_uuid = isset( $candidate['job_uuid'] ) ? trim( (string) $candidate['job_uuid'] ) : '';
            if ( $id > 0 && '' !== $job_uuid ) {
                return array(
                    'id'       => $id,
                    'job_uuid' => $job_uuid,
                );
            }
        }

        $context = ! empty( $entry['context_payload'] ) && is_array( $entry['context_payload'] ) ? $entry['context_payload'] : array();
        $id      = isset( $context['scheduled_self_wake_job_id'] ) ? intval( $context['scheduled_self_wake_job_id'] ) : 0;
        $uuid    = isset( $context['scheduled_self_wake_job_uuid'] ) ? trim( (string) $context['scheduled_self_wake_job_uuid'] ) : '';
        if ( $id > 0 && '' !== $uuid ) {
            return array(
                'id'       => $id,
                'job_uuid' => $uuid,
            );
        }

        return array();
    }

    private function actor_entry_scheduled_attempt_matches( $entry, $expected ) {
        $expected = trim( (string) $expected );
        if ( '' === $expected ) {
            return true;
        }

        $entry      = is_array( $entry ) ? $entry : array();
        $candidates = array();
        if ( ! empty( $entry['message']['scheduled_job']['attempt_token'] ) ) {
            $candidates[] = (string) $entry['message']['scheduled_job']['attempt_token'];
        }
        if ( ! empty( $entry['context_payload']['scheduled_actor_attempt_token'] ) ) {
            $candidates[] = (string) $entry['context_payload']['scheduled_actor_attempt_token'];
        }
        if ( ! empty( $entry['continuation_context']['scheduled_actor_attempt_token'] ) ) {
            $candidates[] = (string) $entry['continuation_context']['scheduled_actor_attempt_token'];
        }

        foreach ( $candidates as $candidate ) {
            if ( hash_equals( $expected, trim( (string) $candidate ) ) ) {
                return true;
            }
        }

        return false;
    }

    private function actor_message_is_superseded( $state, $message_id ) {
        $message_id = trim( (string) $message_id );
        if ( '' === $message_id || ! is_array( $state ) ) {
            return false;
        }

        $ids = isset( $state['superseded_message_ids'] ) && is_array( $state['superseded_message_ids'] ) ? $state['superseded_message_ids'] : array();
        return array_key_exists( $message_id, $ids );
    }

    private function remember_recent_completed_actor_message( $state, $entry ) {
        $state = $this->normalize_actor_state(
            isset( $state['session_id'] ) ? (string) $state['session_id'] : '',
            $state
        );

        if ( ! $this->actor_entry_is_external_founder_message( $entry ) ) {
            return $state;
        }

        $identity = $this->actor_entry_founder_message_identity_key( $entry );
        $fingerprint = $this->actor_entry_founder_message_text_fingerprint_key( $entry );
        if ( '' === $identity && '' === $fingerprint ) {
            return $state;
        }

        $now = time();
        foreach ( array_filter( array( $identity, $fingerprint ) ) as $key ) {
            $state['recent_completed_message_identities'][ $key ] = array(
                'actor_id'     => isset( $entry['id'] ) ? (string) $entry['id'] : '',
                'completed_at' => $now,
            );
        }
        $state['recent_completed_message_identities'] = $this->prune_recent_completed_message_identities(
            $state['recent_completed_message_identities'],
            $now
        );

        return $state;
    }

    private function actor_entry_is_external_founder_message( $entry ) {
        $entry   = is_array( $entry ) ? $entry : array();
        $message = isset( $entry['message'] ) && is_array( $entry['message'] ) ? $entry['message'] : array();

        if ( $this->actor_entry_is_durable_system_message( $entry ) ) {
            return false;
        }

        if ( ! empty( $entry['continuation_context'] ) ) {
            return false;
        }

        $text = isset( $message['text'] ) ? trim( (string) $message['text'] ) : '';
        if ( '[continuation]' === $text ) {
            return false;
        }

        return true;
    }

    private function supersede_existing_actor_work_explicitly( $state ) {
        $state = $this->normalize_actor_state(
            isset( $state['session_id'] ) ? (string) $state['session_id'] : '',
            $state
        );
        $now   = time();
        $count = 0;

        $kept_pending = array();
        foreach ( $state['pending_messages'] as $pending ) {
            if ( $this->actor_entry_can_be_superseded_by_founder_message( $pending ) && ! empty( $pending['id'] ) ) {
                $state['superseded_message_ids'][ (string) $pending['id'] ] = $now;
                $count++;
            } else {
                $kept_pending[] = $pending;
            }
        }
        $state['pending_messages'] = $kept_pending;

        if ( ! empty( $state['active_message']['id'] ) && $this->actor_entry_can_be_superseded_by_founder_message( $state['active_message'] ) ) {
            $state['superseded_message_ids'][ (string) $state['active_message']['id'] ] = $now;
            $state['active_message'] = array();
            $count++;
        }

        if ( $count > 0 ) {
            $state['cancelled_at']      = $now;
            $state['cancel_reason']     = 'Explicit founder supersede requested.';
            $state['cancel_generation'] = intval( isset( $state['cancel_generation'] ) ? $state['cancel_generation'] : 0 ) + 1;
            $state['superseded_message_ids'] = $this->prune_superseded_message_ids( $state['superseded_message_ids'], $now );
        }

        return $state;
    }

	private function actor_entry_can_be_superseded_by_founder_message( $entry ) {
		return ! $this->actor_entry_is_durable_system_message( $entry );
	}

	private function prioritize_external_founder_actor_messages( $messages ) {
		if ( ! is_array( $messages ) || count( $messages ) < 2 ) {
			return is_array( $messages ) ? $messages : array();
		}

		$founder = array();
		$other   = array();
		foreach ( $messages as $entry ) {
			if ( is_array( $entry ) && $this->actor_entry_is_external_founder_message( $entry ) ) {
				$founder[] = $entry;
			} else {
				$other[] = $entry;
			}
		}

		return array_merge( $founder, $other );
	}

	private function requeue_expired_active_actor_message( $state, $active, $now, $reason ) {
		$state     = $this->normalize_actor_state(
			isset( $state['session_id'] ) ? (string) $state['session_id'] : '',
			$state
        );
        $active    = is_array( $active ) ? $active : array();
        $active_id = isset( $active['id'] ) ? (string) $active['id'] : '';
        $attempts  = intval( isset( $active['orphan_recovery_attempts'] ) ? $active['orphan_recovery_attempts'] : 0 );

        if ( $this->actor_entry_has_recent_completed_identity( $state, $active ) ) {
            return $this->discard_active_actor_message_as_completed_duplicate(
                $state,
                $active,
                $now,
                'Actor message lease recovery skipped because the same founder message already completed.'
            );
        }

		if ( $this->actor_entry_is_bound_admin_ui_message( $active ) ) {
			return $this->defer_bound_admin_ui_failure_recovery( $state, $active, $now, $reason );
		}

        unset( $active['claimed_at'], $active['lease_expires_at'], $active['cancel_generation'], $active['last_seen_at'] );
        $active['id']                       = $this->generate_actor_message_id();
        $active['enqueued_at']              = intval( $now );
        $active['orphan_recovery_attempts'] = $attempts + 1;
        $active['orphaned_from_message_id'] = $active_id;
        $active['orphan_recovery_reason']   = sanitize_text_field( (string) $reason );
		$active['not_before']               = intval( $now ) + min( 300, 5 * ( 2 ** min( 6, $attempts ) ) );
        $active                             = $this->stamp_actor_message_current_runtime( $active );

        array_unshift( $state['pending_messages'], $active );
        $state['pending_messages'] = $this->trim_actor_pending_messages( $state['pending_messages'] );
        if ( '' !== $active_id ) {
            $state['superseded_message_ids'][ $active_id ] = intval( $now );
        }
        $state['active_message']     = array();
        $state['cancelled_at']       = 0;
        $state['cancel_reason']      = '';
        $state['cancel_generation']  = intval( isset( $state['cancel_generation'] ) ? $state['cancel_generation'] : 0 ) + 1;
        $state['updated_at']         = intval( $now );
        $state['superseded_message_ids'] = $this->prune_superseded_message_ids( $state['superseded_message_ids'], intval( $now ) );

        return $state;
    }

    private function actor_message_runtime_is_current( $entry ) {
        if ( ! defined( 'NIWA_AI_VERSION' ) ) {
            return true;
        }

        $entry = is_array( $entry ) ? $entry : array();
        $runtime_version = isset( $entry['runtime_version'] ) ? trim( (string) $entry['runtime_version'] ) : '';

        if ( '' !== $runtime_version && $runtime_version === (string) NIWA_AI_VERSION ) {
            return true;
        }

        return $this->actor_message_can_cross_runtime_change( $entry );
    }

    private function actor_message_can_cross_runtime_change( $entry ) {
        $entry = is_array( $entry ) ? $entry : array();

        if ( ! empty( $entry['orphaned_from_message_id'] ) || isset( $entry['orphan_recovery_attempts'] ) ) {
            return true;
        }

        if ( ! empty( $entry['continuation_context'] ) && is_array( $entry['continuation_context'] ) ) {
            return true;
        }

        if ( $this->actor_entry_is_durable_system_message( $entry ) ) {
            return true;
        }

        $message = isset( $entry['message'] ) && is_array( $entry['message'] ) ? $entry['message'] : array();
        if ( empty( $message ) || ! $this->actor_entry_is_external_founder_message( $entry ) ) {
            return false;
        }

        foreach ( array( 'text', 'caption', 'chat_id', 'message_id', 'update_id', 'kind' ) as $key ) {
            if ( isset( $message[ $key ] ) && '' !== trim( (string) $message[ $key ] ) ) {
                return true;
            }
        }

        foreach ( array( 'media_refs', 'reply_media_refs', 'album_message_ids', 'fragment_message_ids' ) as $key ) {
            if ( ! empty( $message[ $key ] ) ) {
                return true;
            }
        }

        return false;
    }

    private function stamp_actor_message_current_runtime( $entry ) {
        $entry = is_array( $entry ) ? $entry : array();
        if ( ! defined( 'NIWA_AI_VERSION' ) ) {
            return $entry;
        }

        $current         = (string) NIWA_AI_VERSION;
        $runtime_version = isset( $entry['runtime_version'] ) ? trim( (string) $entry['runtime_version'] ) : '';
        if ( $runtime_version !== $current ) {
            if ( '' !== $runtime_version ) {
                $entry['runtime_version_migrated_from'] = $runtime_version;
            }
            $entry['runtime_version'] = $current;
        }

        return $entry;
    }

    private function actor_entry_is_scheduled_self_wake_work( $entry ) {
        $entry   = is_array( $entry ) ? $entry : array();
        $message = isset( $entry['message'] ) && is_array( $entry['message'] ) ? $entry['message'] : array();

        $source = '';
        if ( ! empty( $entry['source'] ) ) {
            $source = sanitize_key( (string) $entry['source'] );
        } elseif ( ! empty( $message['source'] ) ) {
            $source = sanitize_key( (string) $message['source'] );
        } elseif ( ! empty( $entry['context_payload']['source'] ) ) {
            $source = sanitize_key( (string) $entry['context_payload']['source'] );
        }

        if ( 'scheduled_self_wake' === $source ) {
            return true;
        }

        if ( ! empty( $entry['context_payload']['scheduled_self_wake_execution'] ) ) {
            return true;
        }

        return ! empty( $this->actor_entry_scheduled_job_reference( $entry ) );
    }

    private function actor_entry_is_durable_system_message( $entry ) {
        $entry   = is_array( $entry ) ? $entry : array();
        $message = isset( $entry['message'] ) && is_array( $entry['message'] ) ? $entry['message'] : array();

        if ( ! empty( $entry['durable'] ) ) {
            return true;
        }

        if ( $this->actor_entry_is_scheduled_self_wake_work( $entry ) ) {
            return true;
        }

        $source = '';
        if ( ! empty( $entry['source'] ) ) {
            $source = sanitize_key( (string) $entry['source'] );
        } elseif ( ! empty( $message['source'] ) ) {
            $source = sanitize_key( (string) $message['source'] );
        }
        if ( '' !== $source && in_array( $source, array( 'imap_idle_listener', 'subagent_result' ), true ) ) {
            return true;
        }

        $context = isset( $entry['context_payload'] ) && is_array( $entry['context_payload'] ) ? $entry['context_payload'] : array();
        return ! empty( $context['inbound_email'] ) && is_array( $context['inbound_email'] );
    }

    private function actor_entry_is_bound_admin_ui_message( $entry ) {
        $entry   = is_array( $entry ) ? $entry : array();
        $message = isset( $entry['message'] ) && is_array( $entry['message'] ) ? $entry['message'] : array();
        $context = isset( $entry['context_payload'] ) && is_array( $entry['context_payload'] ) ? $entry['context_payload'] : array();

        return 'admin_ui' === sanitize_key( (string) ( $message['channel'] ?? '' ) )
            && '' !== trim( (string) ( $context['admin_ui_job_id'] ?? '' ) );
    }

    private function defer_bound_admin_ui_failure_recovery( $state, $active, $now, $reason ) {
        $active   = is_array( $active ) ? $active : array();
        $attempts = max( 1, intval( $active['admin_ui_failure_attempts'] ?? 0 ) + 1 );
        unset( $active['claimed_at'], $active['lease_expires_at'], $active['cancel_generation'], $active['last_seen_at'], $active['worker_pid'] );
        $active['admin_ui_failure_pending']  = true;
        $active['admin_ui_failure_attempts'] = $attempts;
        $active['admin_ui_failure_reason']   = sanitize_text_field( (string) $reason );
        $active['enqueued_at']               = intval( $now );
        $active['not_before']                = intval( $now ) + min( 30, $attempts * 5 );

        array_unshift( $state['pending_messages'], $active );
        $state['pending_messages'] = $this->trim_actor_pending_messages( $state['pending_messages'] );
        $state['active_message']   = array();
        $state['cancelled_at']     = 0;
        $state['cancel_reason']    = '';
        $state['updated_at']       = intval( $now );
        return $state;
    }

    private function trim_actor_pending_messages( $messages ) {
        $messages = array_values( (array) $messages );
        $max      = self::MAX_PENDING_ACTOR_MESSAGES;
        if ( count( $messages ) <= $max ) {
            return $messages;
        }

        $keep = array();
        foreach ( $messages as $index => $entry ) {
            if (
                ! $this->actor_entry_can_be_superseded_by_founder_message( $entry )
                || $this->actor_entry_is_bound_admin_ui_message( $entry )
            ) {
                $keep[ $index ] = true;
            }
        }

        $remaining = $max - count( $keep );
        for ( $index = count( $messages ) - 1; $index >= 0 && $remaining > 0; $index-- ) {
            if ( empty( $keep[ $index ] ) ) {
                $keep[ $index ] = true;
                $remaining--;
            }
        }

        ksort( $keep );
        $trimmed = array();
        foreach ( array_keys( $keep ) as $index ) {
            $trimmed[] = $messages[ $index ];
        }

        return count( $trimmed ) > $max ? array_slice( $trimmed, -$max ) : $trimmed;
    }

    private function normalize_superseded_message_ids( $ids ) {
        $clean = array();
        foreach ( (array) $ids as $id => $timestamp ) {
            $id = trim( (string) $id );
            if ( '' !== $id ) {
                $clean[ $id ] = intval( $timestamp );
            }
        }

        return $this->prune_superseded_message_ids( $clean, time() );
    }

    private function prune_superseded_message_ids( $ids, $now ) {
        $clean = array();
        $cutoff = intval( $now ) - ( defined( 'DAY_IN_SECONDS' ) ? DAY_IN_SECONDS : 86400 );
        foreach ( (array) $ids as $id => $timestamp ) {
            $timestamp = intval( $timestamp );
            if ( $timestamp >= $cutoff ) {
                $clean[ (string) $id ] = $timestamp;
            }
        }

        return $clean;
    }

    private function normalize_recent_completed_message_identities( $identities ) {
        $clean = array();
        foreach ( (array) $identities as $identity => $record ) {
            $identity = trim( (string) $identity );
            if ( '' === $identity ) {
                continue;
            }

            if ( is_array( $record ) ) {
                $completed_at = intval( isset( $record['completed_at'] ) ? $record['completed_at'] : 0 );
                $actor_id     = isset( $record['actor_id'] ) ? sanitize_text_field( (string) $record['actor_id'] ) : '';
            } else {
                $completed_at = intval( $record );
                $actor_id     = '';
            }

            if ( $completed_at > 0 ) {
                $clean[ $identity ] = array(
                    'actor_id'     => $actor_id,
                    'completed_at' => $completed_at,
                );
            }
        }

        return $this->prune_recent_completed_message_identities( $clean, time() );
    }

    private function prune_recent_completed_message_identities( $identities, $now ) {
        $cutoff = intval( $now ) - self::RECENT_COMPLETED_FOUNDER_MESSAGE_TTL;
        $clean  = array();
        foreach ( (array) $identities as $identity => $record ) {
            $identity     = trim( (string) $identity );
            $record       = is_array( $record ) ? $record : array( 'completed_at' => intval( $record ) );
            $completed_at = intval( isset( $record['completed_at'] ) ? $record['completed_at'] : 0 );
            if ( '' === $identity || $completed_at < $cutoff ) {
                continue;
            }

            $clean[ $identity ] = array(
                'actor_id'     => isset( $record['actor_id'] ) ? sanitize_text_field( (string) $record['actor_id'] ) : '',
                'completed_at' => $completed_at,
            );
        }

        uasort(
            $clean,
            function ( $a, $b ) {
                return intval( isset( $b['completed_at'] ) ? $b['completed_at'] : 0 ) <=> intval( isset( $a['completed_at'] ) ? $a['completed_at'] : 0 );
            }
        );

        return array_slice( $clean, 0, self::MAX_RECENT_COMPLETED_FOUNDER_MESSAGES, true );
    }

    private function compact_actor_message( $message ) {
        $message = is_array( $message ) ? $message : array();
        $allowed = array(
            'channel',
            'update_id',
            'chat_id',
            'message_id',
            'text',
            'caption',
            'language_code',
            'remember_as',
            'user_role',
            'source',
            // Phase 2-6 structured payload (Phase 7 source of truth).
            'media_refs',
            'reply_media_refs',
            'media_group_id',
            'kind',
            'placeholder',
            'forward',
            'location',
            'contact',
            'is_edited',
            'is_album',
            'album_size',
            'album_message_ids',
            'is_text_fragment',
            'fragment_count',
            'fragment_message_ids',
            // Skill / spreadsheet plumbing populated by handle_incoming_message.
            'skill_import_content',
            'skill_import_archive_base64',
            'skill_import_filename',
            'spreadsheet_file_id',
            'spreadsheet_filename',
            // Callback queries.
            'callback_query_id',
            'callback_data',
            'callback_origin_message_id',
            // Identity (Phase 1).
            'from_id',
            'from_me',
            'remote_jid',
            'phone_jid',
            'chat_jid',
            'sender_jid',
            'from_username',
            'from_first_name',
            'raw_message_id',
        );
        $payload = array();
        foreach ( $allowed as $key ) {
            if ( array_key_exists( $key, $message ) ) {
                $payload[ $key ] = $message[ $key ];
            }
        }

        return $payload;
    }

    private function generate_actor_message_id() {
        if ( function_exists( 'wp_generate_uuid4' ) ) {
            return 'actor_' . wp_generate_uuid4();
        }

        return 'actor_' . md5( uniqid( 'niwa_actor_', true ) );
    }

    private function wait_for_session_lane( $session_id, $ttl = 20, $attempts = 8 ) {
        $attempts = max( 1, intval( $attempts ) );
        for ( $attempt = 0; $attempt < $attempts; $attempt++ ) {
            if ( $this->acquire_session_lane( $session_id, $ttl ) ) {
                return true;
            }
            usleep( 100000 );
        }

        return false;
    }

    private function session_lock_name( $session_id ) {
        return 'session-' . md5( (string) $session_id );
    }
}
