<?php
/**
 * Recovers actor runtime state after dead runners or stuck stream locks.
 *
 * @package Niwa_AI
 */

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

class Niwa_AI_Admin_Autonomy_Actor_Runtime_Recovery {

    private $telegram;
    private $control_plane;
    private $session_state;
    private $callbacks;

    public function __construct( $telegram, $control_plane, $session_state = null, $callbacks = array() ) {
        $this->telegram      = $telegram;
        $this->control_plane = $control_plane;
        $this->session_state = $session_state;
        $this->callbacks     = is_array( $callbacks ) ? $callbacks : array();
    }

    public function reap_dead_actor_runtime_state() {
        if (
            ! $this->telegram
            || ! $this->control_plane
            || ! method_exists( $this->control_plane, 'recover_orphaned_active_actor_messages' )
            || ! method_exists( $this->telegram, 'get_runner_state' )
        ) {
            return 0;
        }

        $idle_closed_turns = $this->finish_idle_running_turns();
        $state            = (array) $this->telegram->get_runner_state();
        $now              = time();
        $active_until     = intval( isset( $state['active_until'] ) ? $state['active_until'] : 0 );
        $active_runner_id = isset( $state['active_runner_id'] ) ? trim( (string) $state['active_runner_id'] ) : '';
        $last_seen_at     = intval( isset( $state['last_seen_at'] ) ? $state['last_seen_at'] : 0 );
        $last_spawned_at  = intval( isset( $state['last_spawned_at'] ) ? $state['last_spawned_at'] : 0 );
        $last_error       = isset( $state['last_error'] ) ? trim( (string) $state['last_error'] ) : '';
        $active_transport = isset( $state['active_transport'] ) ? sanitize_key( (string) $state['active_transport'] ) : '';
        $active_pid       = intval( isset( $state['active_pid'] ) ? $state['active_pid'] : 0 );
        $stale_runner_id  = '' !== $active_runner_id && $active_until <= $now;
        $explicit_death_trace = preg_match( '/^(died_at|fatal:)/', $last_error );
        $dead_local_runner = false;
        if (
            'local_cli' === $active_transport
            && $active_pid > 0
            && class_exists( 'Niwa_AI_Admin_Autonomy_Process_Runtime' )
            && ! Niwa_AI_Admin_Autonomy_Process_Runtime::background_process_is_alive( $active_pid )
        ) {
            $dead_local_runner = true;
            $explicit_death_trace = true;
            $state['last_error'] = 'died_at:local_cli_pid_' . $active_pid;
        }

        if ( $active_until > $now && ! $dead_local_runner ) {
            return $idle_closed_turns;
        }
        if ( ! $explicit_death_trace && $last_seen_at > 0 && $last_seen_at > ( $now - 60 ) ) {
            return $idle_closed_turns;
        }
        // Spawn-grace window: a runner that was just spawned may still be
        // booting (PHP startup, autoload, WP bootstrap) before it writes its
        // first last_seen_at heartbeat. Defer recovery for up to 30s after the
        // spawn so a slow boot is not mistaken for a silent death. This only
        // ever defers — it never triggers recovery — and is skipped entirely
        // when there is an explicit death trace, so it cannot lose messages.
        if (
            ! $explicit_death_trace
            && $last_spawned_at > 0
            && $last_spawned_at > ( $now - 30 )
            && ( 0 === $last_seen_at || $last_seen_at < $last_spawned_at )
        ) {
            return $idle_closed_turns;
        }

        $silent_death_inferred = ( $last_seen_at > 0 && $last_seen_at <= ( $now - 180 ) )
            || ( $stale_runner_id && ( 0 === $last_seen_at || $last_seen_at <= ( $now - 60 ) ) );

        if ( ! $explicit_death_trace && ! $silent_death_inferred ) {
            if ( $stale_runner_id ) {
                $state['active_runner_id'] = '';
                $state['active_pid']       = 0;
                $this->telegram->update_runner_state( $state );
            }

            return $idle_closed_turns;
        }

        $session_scan = $this->scan_affected_actor_sessions( $now, ! empty( $explicit_death_trace ) );
        $affected_sessions  = $session_scan['affected'];
        $live_lease_sessions = $session_scan['live_lease'];
        if ( empty( $affected_sessions ) ) {
            if ( ! empty( $live_lease_sessions ) ) {
                $this->log(
                    'actor_runtime_recovery_deferred_active_lease',
                    array(
                        'sessions'        => $live_lease_sessions,
                        'last_seen_at'    => $last_seen_at,
                        'last_error'      => $last_error,
                        'stale_runner_id' => $stale_runner_id,
                    )
                );
            } else {
                $state_changed = false;
                if ( $stale_runner_id || $dead_local_runner ) {
                    $state['active_runner_id'] = '';
                    $state['active_until']     = 0;
                    $state['active_pid']       = 0;
                    $state_changed             = true;
                }
                if ( $explicit_death_trace && '' !== trim( (string) $state['last_error'] ) ) {
                    // The death trace is fully consumed: no session was affected
                    // and nothing is deferred behind a live lease. Clear it so a
                    // stale died_at/fatal marker cannot re-arm live-lease
                    // recovery on every later tick.
                    $state['last_error'] = '';
                    $state_changed       = true;
                }
                if ( $state_changed ) {
                    $this->telegram->update_runner_state( $state );
                }
            }

            return $idle_closed_turns;
        }

        $recovered = (array) $this->control_plane->recover_orphaned_active_actor_messages(
            'Actor runtime died without an active runner.',
            50,
            1,
            ! empty( $explicit_death_trace )
        );
        $requeued     = intval( isset( $recovered['requeued'] ) ? $recovered['requeued'] : 0 );
        $cancelled    = intval( isset( $recovered['cancelled'] ) ? $recovered['cancelled'] : 0 );
        $skipped_live = intval( isset( $recovered['skipped_live'] ) ? $recovered['skipped_live'] : 0 );
        $sessions     = ! empty( $recovered['sessions'] ) && is_array( $recovered['sessions'] )
            ? array_values( array_unique( array_map( 'strval', $recovered['sessions'] ) ) )
            : $affected_sessions;

        if ( $requeued > 0 || $cancelled > 0 ) {
            $closed_turns = $this->finish_stale_running_turns( $sessions );
            $state['last_error'] = '';
            if ( $stale_runner_id || $dead_local_runner ) {
                $state['active_runner_id'] = '';
                $state['active_until']     = 0;
                $state['active_pid']       = 0;
            }
            if ( $requeued > 0 ) {
                $state['wake_pending']      = 1;
                $state['wake_reason']       = 'actor-recovery';
                $state['wake_requested_at'] = $now;
            }
            $this->telegram->update_runner_state( $state );
            $this->log(
                'actor_runtime_dead_state_recovered',
                array(
                    'requeued'        => $requeued,
                    'cancelled'       => $cancelled,
                    'skipped_live'     => $skipped_live,
                    'closed_turns'     => $closed_turns,
                    'sessions'         => $sessions,
                    'last_seen_at'     => $last_seen_at,
                    'last_error'       => $last_error,
                    'stale_runner_id'  => $stale_runner_id,
                )
            );
        } elseif ( $stale_runner_id || $dead_local_runner || ( $explicit_death_trace && '' !== trim( (string) $state['last_error'] ) ) ) {
            $state['active_runner_id'] = '';
            $state['active_until']     = 0;
            $state['active_pid']       = 0;
            // Recovery ran to completion without requeueing anything; the
            // death trace is consumed and must not re-arm future sweeps.
            $state['last_error'] = '';
            $this->telegram->update_runner_state( $state );
        }

        return $requeued + $cancelled + $idle_closed_turns;
    }

    public function recover_stuck_turn_streams( $stale_after_seconds = 90, $limit = 20 ) {
        global $wpdb;

        $result = array(
            'success'   => true,
            'scanned'   => 0,
            'recovered' => 0,
            'sessions'  => array(),
            'errors'    => array(),
        );

        if ( ! isset( $wpdb ) || ! is_object( $wpdb ) ) {
            return $result;
        }

        $stale_after_seconds = max( 30, intval( $stale_after_seconds ) );
        $limit               = max( 1, min( 100, intval( $limit ) ) );
        $now                 = time();
        $cutoff              = $now - $stale_after_seconds;

        $rows = $wpdb->get_results(
            $wpdb->prepare(
                "SELECT option_name, option_value FROM {$wpdb->options} WHERE option_name LIKE %s LIMIT %d",
                'niwa_ai_admin_turn_stream_%',
                $limit * 4
            )
        );
        if ( empty( $rows ) || ! is_array( $rows ) ) {
            return $result;
        }

        foreach ( $rows as $row ) {
            if ( $result['recovered'] >= $limit ) {
                break;
            }
            $result['scanned']++;

            $option_name = isset( $row->option_name ) ? (string) $row->option_name : '';
            if ( '' === $option_name ) {
                continue;
            }

            $wrapped = maybe_unserialize( $row->option_value );
            if ( ! is_array( $wrapped ) || ! isset( $wrapped['d'] ) || ! is_array( $wrapped['d'] ) ) {
                delete_option( $option_name );
                continue;
            }

            $state      = (array) $wrapped['d'];
            $expiration = isset( $wrapped['e'] ) ? intval( $wrapped['e'] ) : 0;
            if ( $expiration > 0 && $expiration <= $now ) {
                delete_option( $option_name );
                continue;
            }
            if ( empty( $state['assistant_stream_locked'] ) ) {
                continue;
            }

            $last_event_at = intval( isset( $state['last_event_at'] ) ? $state['last_event_at'] : 0 );
            $started_at    = intval( isset( $state['started_at'] ) ? $state['started_at'] : 0 );
            $reference_at  = $last_event_at > 0 ? $last_event_at : $started_at;
            if ( $reference_at <= 0 || $reference_at >= $cutoff ) {
                continue;
            }

            $this->log(
                'turn_stream_recovery_attempt',
                array(
                    'option_name'   => $option_name,
                    'chat_id'       => isset( $state['chat_id'] ) ? trim( (string) $state['chat_id'] ) : '',
                    'message_id'    => intval( isset( $state['message_id'] ) ? $state['message_id'] : 0 ),
                    'last_event_at' => $last_event_at,
                    'started_at'    => $started_at,
                    'sent_any'      => ! empty( $state['sent_any'] ),
                    'stale_seconds' => $reference_at > 0 ? max( 0, $now - $reference_at ) : -1,
                )
            );

            delete_option( $option_name );
            $result['recovered']++;
        }

        if ( $result['recovered'] > 0 && $this->control_plane && method_exists( $this->control_plane, 'recover_orphaned_active_actor_messages' ) ) {
            try {
                $recovered = (array) $this->control_plane->recover_orphaned_active_actor_messages(
                    'Stuck stream lock recovered by background tick after silent worker death.',
                    50,
                    1
                );
                if ( ! empty( $recovered['sessions'] ) && is_array( $recovered['sessions'] ) ) {
                    $result['sessions'] = array_values( array_unique( array_map( 'strval', $recovered['sessions'] ) ) );
                }
            } catch ( Throwable $e ) {
                $result['errors'][] = 'recover_orphaned: ' . $e->getMessage();
            }
        }

        if ( $result['recovered'] > 0 ) {
            $this->log(
                'turn_stream_recovery_finished',
                array(
                    'scanned'   => $result['scanned'],
                    'recovered' => $result['recovered'],
                    'sessions'  => $result['sessions'],
                )
            );
        }

        return $result;
    }

    private function finish_idle_running_turns() {
        if ( ! $this->session_state || ! method_exists( $this->session_state, 'finish_dead_running_turns_for_idle_actor_sessions' ) ) {
            return 0;
        }

        $closed = intval(
            $this->session_state->finish_dead_running_turns_for_idle_actor_sessions(
                60,
                'cancelled',
                'Actor runtime is idle with no runner, lease, active message, or queue; previous running turn will not resume.'
            )
        );
        if ( $closed > 0 ) {
            $this->log( 'actor_runtime_idle_running_turns_reaped', array( 'closed_turns' => $closed ) );
        }

        return $closed;
    }

    private function scan_affected_actor_sessions( $now, $explicit_death_trace ) {
        $affected = array();
        $live     = array();
        if ( ! method_exists( $this->control_plane, 'list_sessions_with_pending_actor_messages' ) ) {
            return array( 'affected' => $affected, 'live_lease' => $live );
        }

        foreach ( (array) $this->control_plane->list_sessions_with_pending_actor_messages( 10 ) as $record ) {
            $session_id = isset( $record['session_id'] ) ? trim( (string) $record['session_id'] ) : '';
            if ( '' === $session_id || empty( $record['has_active'] ) ) {
                continue;
            }

            $lease_expires_at = intval( isset( $record['active_lease_expires_at'] ) ? $record['active_lease_expires_at'] : 0 );
            $active_last_seen = intval( isset( $record['active_last_seen_at'] ) ? $record['active_last_seen_at'] : 0 );
            $active_worker_pid = intval( isset( $record['active_worker_pid'] ) ? $record['active_worker_pid'] : 0 );

            // Deliberately runtime-version-agnostic: after a plugin deploy the
            // old-runtime worker holding this message is still a healthy
            // process finishing its turn. Death is detected only via a dead
            // worker pid, a long-silent lease, or lease expiry.
            if ( $active_worker_pid > 0 && class_exists( 'Niwa_AI_Admin_Autonomy_Process_Runtime' ) ) {
                if ( ! Niwa_AI_Admin_Autonomy_Process_Runtime::background_process_is_alive( $active_worker_pid ) ) {
                    $affected[] = $session_id;
                } else {
                    // The worker holding this message is provably alive. A death
                    // trace on the loopback-runner state proves only that THAT
                    // runner died — a different live worker (e.g. the internal
                    // clock runner) must never have its in-flight turn reaped.
                    $live[] = $session_id;
                }
                continue;
            }

            if (
                0 === $active_worker_pid
                && $active_last_seen > 0
                && $active_last_seen <= ( $now - 180 )
            ) {
                $affected[] = $session_id;
                continue;
            }

            if ( ! $explicit_death_trace && $lease_expires_at > $now ) {
                $live[] = $session_id;
                continue;
            }

            $affected[] = $session_id;
        }

        return array(
            'affected'   => array_values( array_unique( $affected ) ),
            'live_lease' => array_values( array_unique( $live ) ),
        );
    }

    private function finish_stale_running_turns( $sessions ) {
        if ( ! $this->session_state || ! method_exists( $this->session_state, 'finish_stale_running_turns' ) ) {
            return 0;
        }

        $closed = 0;
        foreach ( (array) $sessions as $session_id ) {
            $closed += intval(
                $this->session_state->finish_stale_running_turns(
                    $session_id,
                    60,
                    'recovered',
                    'Actor runtime died without an active runner; active actor message was recovered and requeued.'
                )
            );
        }

        return $closed;
    }

    private function log( $event, $context = array() ) {
        Niwa_AI_Admin_Autonomy_Callbacks::call( $this->callbacks, 'log', array( $event, is_array( $context ) ? $context : array() ) );
    }
}
