<?php

namespace classes;

require_once WD . '/classes/xlvask_automation_service.php';
require_once WD . '/classes/xlvask_usage_logs_schema_bootstrap.php';

use Exception;
use objects\xlvask_usage_logs_o;
use Throwable;

/**
 * Durable orchestration and read models for XL Vask invoice-period automation.
 *
 * GET callers use only read methods. Schema bootstrap and state mutation happen
 * explicitly from run/decision/cron entry points.
 */
class xlvask_autopilot_service
{
    public const POLICY_VERSION = 'xlvask-autopilot-v1';
    private const PREVIEW_TTL_SECONDS = 900;

    public static function modeCapabilities(string $mode): array
    {
        return match ($mode) {
            'execute' => ['import' => true, 'persist_plans' => true, 'execute_actions' => true, 'run_artifacts' => true],
            'dry_run' => ['import' => true, 'persist_plans' => true, 'execute_actions' => false, 'run_artifacts' => true],
            'replay' => ['import' => false, 'persist_plans' => false, 'execute_actions' => false, 'run_artifacts' => true],
            default => throw new Exception('Invalid XL Vask autopilot mode.'),
        };
    }

    public static function normalizeHallScope(array $hallIds): array
    {
        $normalized = array_values(array_unique(array_filter(array_map(
            static fn(mixed $id): string => trim((string)$id),
            $hallIds
        ), static fn(string $id): bool => $id !== '' && strlen($id) <= 191)));
        sort($normalized, SORT_STRING);
        return $normalized;
    }

    public static function idempotencyKey(string $providedKey, ?int $actorId, string $nonce): string
    {
        $material = trim($providedKey) !== '' ? trim($providedKey) : $nonce;
        return hash('sha256', self::POLICY_VERSION . ':' . $material . ':' . (string)$actorId);
    }

    public static function requestFingerprint(array $request): string
    {
        return hash('sha256', xlvask_automation_service::stableJsonForAutomation($request));
    }

    public static function previewSnapshotMatches(int $expectedVersion, string $sourceHash, array $current): bool
    {
        return (int)($current['expected_version'] ?? 0) === $expectedVersion
            && $sourceHash !== ''
            && hash_equals($sourceHash, (string)($current['source_hash'] ?? ''));
    }

    public function createRun(array $input, ?int $actorId = null, array $allowedHallIds = []): array
    {
        xlvask_usage_logs_schema_bootstrap::ensureTables();
        global $db;

        $mode = strtolower(trim((string)($input['mode'] ?? 'execute')));
        self::modeCapabilities($mode);
        $dateFrom = $this->normalizeDate($input['dateFrom'] ?? null, 'dateFrom');
        $dateTo = $this->normalizeDate($input['dateTo'] ?? null, 'dateTo');
        if ($dateFrom !== null && $dateTo !== null && $dateFrom > $dateTo) {
            throw new Exception('dateFrom must not be after dateTo.');
        }
        $ids = array_values(array_unique(array_filter(
            array_map('intval', is_array($input['ids'] ?? null) ? $input['ids'] : []),
            static fn(int $id): bool => $id > 0
        )));
        if (count($ids) > 500) {
            throw new Exception('At most 500 XL Vask usage logs can be included in one run.');
        }
        sort($ids, SORT_NUMERIC);
        $requestedLimit = max(1, min(500, (int)($input['limit'] ?? 500)));
        if ($dateFrom !== null && $dateTo !== null) {
            $rangeDays = (int)floor((strtotime($dateTo) - strtotime($dateFrom)) / 86400) + 1;
            $maxRange = $mode === 'replay' ? 366 : 90;
            if ($rangeDays > $maxRange) {
                throw new Exception("XL Vask {$mode} range cannot exceed {$maxRange} days.");
            }
        }
        $forceRefetch = filter_var($input['forceRefetch'] ?? false, FILTER_VALIDATE_BOOL);
        $allowedHallIds = $this->normalizeHallIds($allowedHallIds);
        if ($allowedHallIds === []) {
            throw new Exception('No XL Vask hall scope is available for this user.');
        }
        // An explicit retry key is stable; otherwise every requested rerun is a new durable run.
        $providedKey = trim((string)($input['idempotency_key'] ?? ''));
        $key = self::idempotencyKey($providedKey, $actorId, $this->uuidV4());
        $requestHash = self::requestFingerprint([
            'mode' => $mode,
            'date_from' => $dateFrom,
            'date_to' => $dateTo,
            'ids' => $ids,
            'limit' => $requestedLimit,
            'force_refetch' => $forceRefetch,
            'scope_hall_ids' => $allowedHallIds,
        ]);
        $idsJson = $db->escape_string(json_encode($ids, JSON_UNESCAPED_SLASHES) ?: '[]');
        $scopeJson = $db->escape_string(json_encode($allowedHallIds, JSON_UNESCAPED_SLASHES) ?: '[]');
        $dateFromSql = $dateFrom === null ? 'NULL' : "'" . $db->escape_string($dateFrom) . "'";
        $dateToSql = $dateTo === null ? 'NULL' : "'" . $db->escape_string($dateTo) . "'";
        $actorSql = $actorId === null ? 'NULL' : (string)$actorId;
        $keySql = $db->escape_string($key);
        $modeSql = $db->escape_string($mode);

        $db->query(
            "INSERT INTO xlvask_autopilot_runs
                (idempotency_key, request_hash, mode, date_from, date_to, force_refetch, requested_ids_json,
                 requested_limit, scope_hall_ids_json, created_by)
             VALUES ('{$keySql}', '{$requestHash}', '{$modeSql}', {$dateFromSql}, {$dateToSql}, " . ($forceRefetch ? '1' : '0') . ",
                '{$idsJson}', {$requestedLimit}, '{$scopeJson}', {$actorSql})
             ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id)"
        );
        $runId = (int)$db->insert_id();
        $existing = $db->fetch_assoc($db->query("SELECT request_hash FROM xlvask_autopilot_runs WHERE id = {$runId} LIMIT 1"));
        if (!hash_equals($requestHash, (string)($existing['request_hash'] ?? ''))) {
            throw new Exception('XL Vask idempotency key was already used with a different request payload.');
        }

        return $this->getRun($runId, $actorId, $allowedHallIds);
    }

    public function processQueuedRuns(int $limit = 3): array
    {
        global $db;
        xlvask_usage_logs_schema_bootstrap::ensureTables();
        $limit = max(1, min(10, $limit));
        $db->query(
            "UPDATE xlvask_autopilot_runs
             SET status = 'failed', phase = 'failed', error = 'XL Vask autopilot retry budget exhausted.',
                 lease_token = NULL, lease_expires_at = NULL, finished_at = NOW()
             WHERE status IN ('queued', 'running') AND attempt_count >= max_attempts
               AND (status = 'queued' OR lease_expires_at < NOW())"
        );
        $rows = $db->fetch_all($db->query(
            "SELECT id FROM xlvask_autopilot_runs
             WHERE attempt_count < max_attempts AND (
                    (status = 'queued' AND (next_attempt_at IS NULL OR next_attempt_at <= NOW()))
                    OR (status = 'running' AND lease_expires_at < NOW())
                 )
             ORDER BY id ASC LIMIT {$limit}"
        ));
        $runs = [];
        foreach ($rows as $row) {
            $runs[] = $this->processRun((int)$row['id']);
        }
        return $runs;
    }

    public function processRun(int $runId): array
    {
        global $db;
        xlvask_usage_logs_schema_bootstrap::ensureTables();
        $connection = $db->conn();
        $connection->begin_transaction();
        try {
            $result = $db->query("SELECT * FROM xlvask_autopilot_runs WHERE id = {$runId} FOR UPDATE");
            if ($result === false || $result->num_rows < 1) {
                throw new Exception('XL Vask autopilot run not found.');
            }
            $run = $db->fetch_assoc($result);
            $leaseExpired = $run['lease_expires_at'] !== null && strtotime((string)$run['lease_expires_at']) < time();
            if (in_array((string)$run['status'], ['completed', 'completed_with_warnings', 'failed'], true)
                || ((string)$run['status'] === 'running' && !$leaseExpired)) {
                $connection->commit();
                return $this->formatRun($run);
            }
            $leaseToken = $this->uuidV4();
            $db->query(
                "UPDATE xlvask_autopilot_runs SET status = 'running', phase = 'importing',
                 lease_token = '{$leaseToken}', lease_expires_at = DATE_ADD(NOW(), INTERVAL 15 MINUTE),
                 attempt_count = attempt_count + 1, next_attempt_at = NULL,
                 started_at = COALESCE(started_at, NOW()) WHERE id = {$runId}"
            );
            $run['attempt_count'] = (int)($run['attempt_count'] ?? 0) + 1;
            $connection->commit();
        } catch (Throwable $throwable) {
            $connection->rollback();
            throw $throwable;
        }

        try {
            $dateFrom = $run['date_from'] ?: null;
            $dateTo = $run['date_to'] ?: null;
            $ids = json_decode((string)($run['requested_ids_json'] ?? '[]'), true);
            $ids = is_array($ids) ? array_values(array_filter(array_map('intval', $ids))) : [];
            $scopeHallIds = json_decode((string)($run['scope_hall_ids_json'] ?? '[]'), true);
            $scopeHallIds = is_array($scopeHallIds) ? $this->normalizeHallIds($scopeHallIds) : [];
            $importSummary = ['fetched' => 0, 'upstream_fetched' => 0, 'new' => 0, 'updated' => 0, 'unchanged' => 0, 'invalid' => 0, 'errors' => []];
            $capabilities = self::modeCapabilities((string)$run['mode']);
            $isReplay = !$capabilities['persist_plans'];
            if ($capabilities['import'] && ($ids === [] || (int)$run['force_refetch'] === 1)) {
                $importSummary = (new xlvask_usage_logs_o())->importUsageLogsWithSummary($dateFrom, $dateTo, $scopeHallIds);
            }

            $leaseTokenSql = $db->escape_string($leaseToken);
            $renewLease = static function () use ($db, $runId, $leaseTokenSql): void {
                if ($db->query(
                    "UPDATE xlvask_autopilot_runs
                     SET lease_expires_at = DATE_ADD(NOW(), INTERVAL 15 MINUTE)
                     WHERE id = {$runId} AND status = 'running' AND lease_token = '{$leaseTokenSql}'"
                ) === false) {
                    throw new Exception('XL Vask autopilot lease ownership was lost during processing.');
                }
                if ($db->conn()->affected_rows !== 1) {
                    $owned = $db->query(
                        "SELECT id FROM xlvask_autopilot_runs
                         WHERE id = {$runId} AND status = 'running' AND lease_token = '{$leaseTokenSql}' LIMIT 1"
                    );
                    if ($owned === false || $owned->num_rows !== 1) {
                        throw new Exception('XL Vask autopilot lease ownership was lost during processing.');
                    }
                }
            };
            $db->query("UPDATE xlvask_autopilot_runs SET phase = 'evaluating' WHERE id = {$runId}");
            $renewLease();
            $automation = new xlvask_automation_service();
            $automation->setRunContext($runId);
            $allowExecute = $capabilities['execute_actions'];
            $results = $automation->runPending(
                $dateFrom,
                $dateTo,
                $ids,
                max(1, min(500, (int)($run['requested_limit'] ?? 500))),
                isset($run['created_by']) ? (int)$run['created_by'] : null,
                $allowExecute,
                $runId,
                $scopeHallIds,
                $isReplay,
                $renewLease
            );
            $this->persistRunItems($runId, $results['results'] ?? []);
            $summary = $this->getSummary($dateFrom, $dateTo, $scopeHallIds);
            $summary['import'] = $importSummary;
            $circuitBreaker = $results['circuit_breaker'] ?? null;
            $warning = $circuitBreaker === null ? null : 'Autopilot stopped early: ' . (string)$circuitBreaker;
            $summary['circuit_breaker'] = $circuitBreaker;
            $summaryJson = $db->escape_string(json_encode($summary, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) ?: '{}');
            $processed = (int)($results['processed'] ?? 0);
            $total = (int)($results['eligible_total'] ?? $results['selected'] ?? $processed);
            if ($circuitBreaker === null && $processed < $total) {
                $circuitBreaker = 'batch_limit_reached';
                $warning = 'Autopilot stopped early: batch_limit_reached';
                $summary['circuit_breaker'] = $circuitBreaker;
                $summaryJson = $db->escape_string(json_encode($summary, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) ?: '{}');
            }
            $status = $circuitBreaker === null ? 'completed' : 'completed_with_warnings';
            $phase = $circuitBreaker === null ? 'completed' : 'circuit_breaker';
            $warningSql = $warning === null ? 'NULL' : "'" . $db->escape_string($warning) . "'";
            $db->query(
                "UPDATE xlvask_autopilot_runs SET status = '{$status}', phase = '{$phase}', processed = {$processed},
                 total = {$total}, summary_json = '{$summaryJson}', warning = {$warningSql}, finished_at = NOW()
                 WHERE id = {$runId} AND lease_token = '" . $db->escape_string($leaseToken) . "'"
            );
            if ($db->conn()->affected_rows !== 1) {
                throw new Exception('XL Vask autopilot lease ownership was lost before completion.');
            }
            if (!$isReplay && $dateFrom !== null && $dateTo !== null) {
                (new redis())->clear_invoice_period_automatic_flags($dateFrom, $dateTo);
            }
        } catch (Throwable $throwable) {
            $message = $db->escape_string('XL Vask autopilot run failed. Review the server-side audit trail.');
            $attempt = (int)($run['attempt_count'] ?? 1);
            $maxAttempts = max(1, (int)($run['max_attempts'] ?? 3));
            if ($attempt < $maxAttempts) {
                $backoffMinutes = min(15, 2 ** max(0, $attempt - 1));
                $db->query(
                    "UPDATE xlvask_autopilot_runs SET status = 'queued', phase = 'retry_wait', error = '{$message}',
                     warning = 'Transient failure; retry scheduled.', lease_token = NULL, lease_expires_at = NULL,
                     next_attempt_at = DATE_ADD(NOW(), INTERVAL {$backoffMinutes} MINUTE)
                     WHERE id = {$runId} AND lease_token = '" . $db->escape_string($leaseToken) . "'"
                );
            } else {
                $db->query(
                    "UPDATE xlvask_autopilot_runs SET status = 'failed', phase = 'failed', error = '{$message}',
                     lease_token = NULL, lease_expires_at = NULL, finished_at = NOW()
                     WHERE id = {$runId} AND lease_token = '" . $db->escape_string($leaseToken) . "'"
                );
            }
        }

        return $this->getRun($runId, isset($run['created_by']) ? (int)$run['created_by'] : null, $scopeHallIds ?? []);
    }

    public function getRun(int $runId, ?int $actorId = null, array $allowedHallIds = []): array
    {
        global $db;
        $result = $db->query("SELECT * FROM xlvask_autopilot_runs WHERE id = {$runId} LIMIT 1");
        if ($result === false || $result->num_rows < 1) {
            throw new Exception('XL Vask autopilot run not found.');
        }
        $row = $db->fetch_assoc($result);
        $runScope = json_decode((string)($row['scope_hall_ids_json'] ?? '[]'), true);
        $runScope = is_array($runScope) ? $this->normalizeHallIds($runScope) : [];
        $allowedHallIds = $this->normalizeHallIds($allowedHallIds);
        if ($actorId !== null && $allowedHallIds === []) {
            throw new Exception('No XL Vask hall scope is available for this user.');
        }
        if ($actorId !== null && (int)($row['created_by'] ?? 0) !== $actorId) {
            throw new Exception('XL Vask autopilot run belongs to another user.');
        }
        if ($allowedHallIds !== [] && array_diff($runScope, $allowedHallIds) !== []) {
            throw new Exception('XL Vask autopilot run is outside the current hall scope.');
        }
        return $this->formatRun($row);
    }

    public function getSummary(?string $dateFrom = null, ?string $dateTo = null, array $allowedHallIds = []): array
    {
        global $db;
        $states = [
            'total', 'new', 'updated', 'unchanged', 'invalid', 'already_linked', 'auto_linked',
            'auto_created', 'needs_review', 'blocked', 'ignored', 'failed', 'certain', 'uncertain', 'none',
        ];
        $summary = array_fill_keys($states, 0);
        $startExpression = "STR_TO_DATE(REPLACE(SUBSTRING(StartTime, 1, 19), 'T', ' '), '%Y-%m-%d %H:%i:%s')";
        $where = ['1=1'];
        $allowedHallIds = $this->normalizeHallIds($allowedHallIds);
        if ($allowedHallIds === []) {
            throw new Exception('No XL Vask hall scope is available for this user.');
        }
        $where[] = 'HallId IN (' . $this->quotedHallIds($allowedHallIds) . ')';
        if ($dateFrom !== null && strtotime($dateFrom) !== false) {
            $where[] = "{$startExpression} >= '" . $db->escape_string(date('Y-m-d 00:00:00', strtotime($dateFrom))) . "'";
        }
        if ($dateTo !== null && strtotime($dateTo) !== false) {
            $where[] = "{$startExpression} <= '" . $db->escape_string(date('Y-m-d 23:59:59', strtotime($dateTo))) . "'";
        }

        try {
            $row = $db->fetch_assoc($db->query(
                "SELECT COUNT(*) total,
                    SUM(import_state = 'new') `new`, SUM(import_state = 'updated') updated,
                    SUM(import_state = 'unchanged') unchanged, SUM(import_state = 'invalid') invalid,
                    SUM(resolution_state = 'already_linked') already_linked,
                    SUM(resolution_state = 'auto_linked') auto_linked,
                    SUM(resolution_state = 'auto_created') auto_created,
                    SUM(resolution_state = 'needs_review') needs_review,
                    SUM(resolution_state = 'blocked') blocked,
                    SUM(resolution_state = 'ignored') ignored,
                    SUM(resolution_state = 'failed') failed,
                    SUM(certainty = 'certain') certain, SUM(certainty = 'uncertain') uncertain,
                    SUM(certainty = 'none') none
                 FROM xlvask_usage_logs WHERE " . implode(' AND ', $where)
            ));
            foreach ($summary as $key => $_) {
                $summary[$key] = (int)($row[$key] ?? 0);
            }
        } catch (Throwable) {
            $row = $db->fetch_assoc($db->query(
                'SELECT COUNT(*) total FROM xlvask_usage_logs WHERE ' . implode(' AND ', $where)
            ));
            $summary['total'] = (int)($row['total'] ?? 0);
            $summary['needs_review'] = $summary['total'];
        }

        return $summary;
    }

    public function activationReadiness(): array
    {
        global $db;
        $active = [];
        try {
            $result = $db->query(
                "SELECT id, policy_version, segment_key, precision_value, wilson_lower_bound,
                        holdout_examples, segment_examples, contradictions, calibrated_probability,
                        artifact_hash, backtest_json, activated_at
                 FROM xlvask_automation_calibrations WHERE active = 1 ORDER BY segment_key"
            );
            if ($result !== false) {
                $verified = [];
                foreach ($db->fetch_all($result) as $row) {
                    $artifact = json_decode((string)($row['backtest_json'] ?? ''), true);
                    if (!is_array($artifact)
                        || ($artifact['split_rule'] ?? null) !== 'chronological_80_20_by_suggestion_created_at_and_id'
                        || ($artifact['policy_version'] ?? null) !== (string)$row['policy_version']
                        || ($artifact['segment_key'] ?? null) !== (string)$row['segment_key']
                        || !hash_equals(
                            (string)($row['artifact_hash'] ?? ''),
                            hash('sha256', xlvask_automation_service::stableJsonForAutomation($artifact))
                        )) {
                        continue;
                    }
                    $verified[] = [
                        ...$artifact,
                        'id' => (int)$row['id'],
                        'active' => true,
                        'artifact_hash' => (string)$row['artifact_hash'],
                        'activated_at' => $row['activated_at'] ?? null,
                    ];
                }
                $active = array_values(array_filter(
                    $verified,
                    static fn(array $artifact): bool => xlvask_automation_service::classifyCertaintyForAutomation($artifact) === 'certain'
                ));
            }
        } catch (Throwable) {
            // Readiness GET is intentionally read-only and fails closed when schema is not ready.
        }
        return [
            'wash_id_uniqueness_ready' => xlvask_usage_logs_schema_bootstrap::washIdUniquenessReady(),
            'active_calibrations' => $active,
            'automatic_actions_ready' => xlvask_usage_logs_schema_bootstrap::washIdUniquenessReady() && $active !== [],
            'wash_id_activation_phrase' => 'ACTIVATE-WASH-ID-UNIQUENESS',
        ];
    }

    public function adjudicateCalibrationLabel(int $suggestionId, string $outcome, ?int $actorId): array
    {
        global $db;
        xlvask_usage_logs_schema_bootstrap::ensureTables();
        if ($suggestionId < 1 || !in_array($outcome, ['correct', 'incorrect'], true) || $actorId === null) {
            throw new Exception('Invalid XL Vask calibration adjudication.');
        }
        $suggestion = $db->query(
            "SELECT id, policy_version, source, action FROM xlvask_automation_suggestions WHERE id = {$suggestionId} LIMIT 1"
        );
        if ($suggestion === false || $suggestion->num_rows < 1) {
            throw new Exception('XL Vask suggestion not found for calibration adjudication.');
        }
        $outcomeSql = $db->escape_string($outcome);
        if ($db->query(
            "INSERT INTO xlvask_automation_calibration_label_events
                (suggestion_id, outcome, adjudicated_by, adjudicated_at)
             VALUES ({$suggestionId}, '{$outcomeSql}', {$actorId}, NOW())"
        ) === false) {
            throw new Exception('XL Vask calibration adjudication could not be stored.');
        }
        return ['suggestion_id' => $suggestionId, 'outcome' => $outcome, 'adjudicated' => true];
    }

    /** Generate an inactive, PII-free artifact from exact admin-adjudicated suggestion labels. */
    public function generateCalibrationArtifact(string $segmentKey, ?int $actorId): array
    {
        global $db;
        xlvask_usage_logs_schema_bootstrap::ensureTables();
        if (!preg_match('/^(deterministic|fuzzy|history):(attach_order|create_order)$/', $segmentKey)) {
            throw new Exception('Invalid XL Vask calibration segment.');
        }
        [$source, $action] = explode(':', $segmentKey, 2);
        $sourceSql = $db->escape_string($source);
        $actionSql = $db->escape_string($action);
        $labels = $db->fetch_all($db->query(
            "SELECT l.id, l.suggestion_id, l.outcome, l.adjudicated_at,
                    s.policy_version, s.source, s.action, s.created_at AS suggestion_created_at
             FROM xlvask_automation_calibration_label_events l
             LEFT JOIN xlvask_automation_calibration_label_events newer
               ON newer.suggestion_id = l.suggestion_id AND newer.id > l.id
             INNER JOIN xlvask_automation_suggestions s ON s.id = l.suggestion_id
             WHERE s.source = '{$sourceSql}' AND s.action = '{$actionSql}'
               AND s.policy_version = '" . self::POLICY_VERSION . "'
               AND newer.id IS NULL
             ORDER BY s.created_at ASC, s.id ASC, l.id ASC"
        ));
        $overallRow = $db->fetch_assoc($db->query(
            "SELECT COUNT(*) AS total
             FROM xlvask_automation_calibration_label_events l
             LEFT JOIN xlvask_automation_calibration_label_events newer
               ON newer.suggestion_id = l.suggestion_id AND newer.id > l.id
             INNER JOIN xlvask_automation_suggestions s ON s.id = l.suggestion_id
             WHERE s.policy_version = '" . self::POLICY_VERSION . "' AND newer.id IS NULL"
        ));
        $overallExamples = (int)($overallRow['total'] ?? 0);
        $total = count($labels);
        $trainingExamples = (int)floor($total * 0.8);
        $holdout = array_slice($labels, $trainingExamples);
        $holdoutExamples = count($holdout);
        $accepted = count(array_filter($holdout, static fn(array $label): bool => $label['outcome'] === 'correct'));
        $denied = count(array_filter($holdout, static fn(array $label): bool => $label['outcome'] === 'incorrect'));
        $contradictions = count(array_filter($labels, static fn(array $label): bool => $label['outcome'] === 'incorrect'));
        $precision = $holdoutExamples > 0 ? $accepted / $holdoutExamples : 0.0;
        $wilson = $this->wilsonLowerBound($accepted, $holdoutExamples);
        $snapshot = array_map(static fn(array $label): array => [
            'id' => (int)$label['id'],
            'suggestion_id' => (int)$label['suggestion_id'],
            'outcome' => (string)$label['outcome'],
            'adjudicated_at' => (string)$label['adjudicated_at'],
            'suggestion_created_at' => (string)$label['suggestion_created_at'],
            'policy_version' => (string)$label['policy_version'],
            'source' => (string)$label['source'],
            'action' => (string)$label['action'],
        ], $labels);
        $artifact = [
            'policy_version' => self::POLICY_VERSION,
            'segment_key' => $segmentKey,
            'split_rule' => 'chronological_80_20_by_suggestion_created_at_and_id',
            'training_examples' => $trainingExamples,
            'holdout_examples' => $holdoutExamples,
            'overall_examples' => $overallExamples,
            'segment_examples' => $total,
            'correct_holdout_examples' => $accepted,
            'incorrect_holdout_examples' => $denied,
            'contradictions' => $contradictions,
            'precision_value' => round($precision, 6),
            'wilson_lower_bound' => round($wilson, 6),
            'calibrated_probability' => round($precision, 6),
            'label_snapshot' => $snapshot,
            'label_snapshot_hash' => hash('sha256', xlvask_automation_service::stableJsonForAutomation($snapshot)),
        ];
        $artifactHash = hash('sha256', xlvask_automation_service::stableJsonForAutomation($artifact));
        $artifactJson = $db->escape_string(xlvask_automation_service::stableJsonForAutomation($artifact));
        $actorSql = $actorId === null ? 'NULL' : (string)$actorId;
        $db->query(
            "INSERT INTO xlvask_automation_calibrations
                (policy_version, segment_key, precision_value, wilson_lower_bound, holdout_examples,
                 segment_examples, contradictions, calibrated_probability, artifact_hash, active, backtest_json, created_by)
             VALUES ('" . self::POLICY_VERSION . "', '" . $db->escape_string($segmentKey) . "', " . round($precision, 6) . ",
                " . round($wilson, 6) . ", {$holdoutExamples}, {$total}, {$contradictions}, " . round($precision, 6) . ",
                '{$artifactHash}', 0, '{$artifactJson}', {$actorSql})
             ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id)"
        );
        $id = (int)$db->insert_id();
        $qualifies = xlvask_automation_service::classifyCertaintyForAutomation([
            ...$artifact,
            'active' => true,
        ]) === 'certain';
        return [
            'id' => $id,
            ...$artifact,
            'artifact_hash' => $artifactHash,
            'active' => false,
            'qualifies_for_activation' => $qualifies,
            'activation_phrase' => 'ACTIVATE-CALIBRATION-' . $id,
        ];
    }

    public function activateCalibration(int $id, string $artifactHash, string $confirmationText, ?int $actorId): array
    {
        global $db;
        xlvask_usage_logs_schema_bootstrap::ensureTables();
        if ($id < 1 || !preg_match('/^[0-9a-f]{64}$/', $artifactHash)
            || !hash_equals('ACTIVATE-CALIBRATION-' . $id, trim($confirmationText))) {
            throw new Exception('Invalid XL Vask calibration activation confirmation.');
        }
        $connection = $db->conn();
        $connection->begin_transaction();
        try {
            $result = $db->query("SELECT * FROM xlvask_automation_calibrations WHERE id = {$id} FOR UPDATE");
            $row = $result !== false && $result->num_rows > 0 ? $db->fetch_assoc($result) : null;
            if ($row === null || !hash_equals((string)$row['artifact_hash'], $artifactHash)) {
                throw new Exception('XL Vask calibration artifact not found or changed.');
            }
            $backtest = json_decode((string)($row['backtest_json'] ?? ''), true);
            if (!is_array($backtest)
                || ($backtest['split_rule'] ?? null) !== 'chronological_80_20_by_suggestion_created_at_and_id'
                || ($backtest['policy_version'] ?? null) !== (string)$row['policy_version']
                || ($backtest['segment_key'] ?? null) !== (string)$row['segment_key']
                || !hash_equals($artifactHash, hash('sha256', xlvask_automation_service::stableJsonForAutomation($backtest)))) {
                throw new Exception('XL Vask calibration artifact payload failed integrity verification.');
            }
            if (xlvask_automation_service::classifyCertaintyForAutomation([...$backtest, 'active' => true]) !== 'certain') {
                throw new Exception('XL Vask calibration artifact does not meet the activation thresholds.');
            }
            $policy = $db->escape_string((string)$row['policy_version']);
            $segment = $db->escape_string((string)$row['segment_key']);
            $db->query(
                "UPDATE xlvask_automation_calibrations SET active = 0
                 WHERE policy_version = '{$policy}' AND segment_key = '{$segment}' AND active = 1"
            );
            $actorSql = $actorId === null ? 'NULL' : (string)$actorId;
            if ($db->query(
                "UPDATE xlvask_automation_calibrations
                 SET active = 1, activated_by = {$actorSql}, activated_at = NOW() WHERE id = {$id} AND active = 0"
            ) === false || $db->conn()->affected_rows !== 1) {
                throw new Exception('XL Vask calibration artifact could not be activated atomically.');
            }
            $connection->commit();
        } catch (Throwable $throwable) {
            $connection->rollback();
            throw $throwable;
        }
        return ['id' => $id, 'active' => true, 'segment_key' => (string)$row['segment_key']];
    }

    public function activateWashIdUniqueness(string $confirmationText): array
    {
        if (!hash_equals('ACTIVATE-WASH-ID-UNIQUENESS', trim($confirmationText))) {
            throw new Exception('Invalid wash-id uniqueness activation confirmation.');
        }
        $ready = xlvask_usage_logs_schema_bootstrap::applyWashIdUniquenessMigration();
        if (!$ready) {
            throw new Exception('Wash-id uniqueness activation is blocked by schema readiness or duplicate wash IDs.');
        }
        return ['wash_id_uniqueness_ready' => true];
    }

    public function createDecisionPreview(array $input, ?int $actorId = null, array $allowedHallIds = []): array
    {
        global $db;
        xlvask_usage_logs_schema_bootstrap::ensureTables();
        $ids = array_values(array_unique(array_filter(array_map(
            'intval',
            is_array($input['usage_log_ids'] ?? null) ? $input['usage_log_ids'] : []
        ))));
        if ($ids === []) {
            throw new Exception('At least one XL Vask usage log is required.');
        }
        if (count($ids) > 100) {
            throw new Exception('At most 100 XL Vask usage logs can be reviewed at once.');
        }
        $action = strtolower(trim((string)($input['action'] ?? '')));
        if (!in_array($action, ['accept', 'deny', 'ignore', 'attach_order', 'create_order'], true)) {
            throw new Exception('Invalid XL Vask review action.');
        }
        $allowedHallIds = $this->normalizeHallIds($allowedHallIds);
        if ($allowedHallIds === []) {
            throw new Exception('No XL Vask hall scope is available for this user.');
        }
        $result = $db->query(
            'SELECT id, WashId, import_state, resolution_state, certainty, planned_action, expected_version, source_hash '
            . 'FROM xlvask_usage_logs WHERE id IN (' . implode(',', $ids) . ') AND HallId IN (' . $this->quotedHallIds($allowedHallIds) . ') ORDER BY id'
        );
        $rows = $db->fetch_all($result);
        if (count($rows) !== count($ids)) {
            throw new Exception('One or more XL Vask usage logs were not found.');
        }
        $items = [];
        foreach ($rows as $row) {
            $usageId = (int)$row['id'];
            if ((string)($row['import_state'] ?? '') === 'invalid') {
                throw new Exception("XL Vask usage log {$usageId} has invalid source data.");
            }
            $suggestion = null;
            if ($action !== 'ignore') {
                $suggestionResult = $db->query(
                    "SELECT id, usage_log_id, action, matched_order_id FROM xlvask_automation_suggestions
                     WHERE usage_log_id = {$usageId} AND status = 'suggested' ORDER BY id DESC LIMIT 1"
                );
                $suggestion = $suggestionResult !== false && $suggestionResult->num_rows > 0
                    ? $db->fetch_assoc($suggestionResult) : null;
                if ($suggestion === null) {
                    throw new Exception("XL Vask usage log {$usageId} has no actionable suggestion.");
                }
                if (isset($input['suggestion_id']) && count($rows) === 1 && (int)$suggestion['id'] !== (int)$input['suggestion_id']) {
                    throw new Exception('The selected XL Vask suggestion is stale.');
                }
                if ($action === 'attach_order' && (int)($input['order_id'] ?? 0) > 0) {
                    $candidate = (new xlvask_automation_service())->validateManualCandidate(
                        $usageId,
                        (int)$input['order_id']
                    );
                    $suggestion['matched_order_id'] = (int)$candidate['id'];
                }
            }
            $items[] = [
                'usage_log_id' => $usageId,
                'suggestion_id' => $suggestion === null ? null : (int)$suggestion['id'],
                'candidate_order_id' => $suggestion === null || $suggestion['matched_order_id'] === null
                    ? null : (int)$suggestion['matched_order_id'],
                'suggested_action' => $suggestion['action'] ?? null,
                'before' => [
                    'resolution_state' => (string)$row['resolution_state'],
                    'certainty' => (string)$row['certainty'],
                    'planned_action' => (string)$row['planned_action'],
                    'expected_version' => (int)$row['expected_version'],
                    'source_hash' => (string)$row['source_hash'],
                ],
                'after' => ['action' => $action],
                'warnings' => [],
            ];
        }
        $requiresConfirmation = count($items) > 1 || in_array($action, ['deny', 'ignore'], true);
        $confirmationPhrase = $requiresConfirmation ? 'CONFIRM-' . strtoupper(bin2hex(random_bytes(4))) : null;
        $payload = [
            'action' => $action,
            'items' => $items,
            'reason' => mb_substr(trim((string)($input['reason'] ?? '')), 0, 1000),
            'confirmation_phrase' => $confirmationPhrase,
        ];
        $selectionHash = hash('sha256', xlvask_automation_service::stableJsonForAutomation($payload));
        $id = $this->uuidV4();
        $expiresAt = date('Y-m-d H:i:s', time() + self::PREVIEW_TTL_SECONDS);
        $payloadJson = $db->escape_string(json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) ?: '{}');
        $actorSql = $actorId === null ? 'NULL' : (string)$actorId;
        $db->query(
            "INSERT INTO xlvask_automation_decision_previews
                (id, selection_hash, action, payload_json, created_by, expires_at)
             VALUES ('{$id}', '{$selectionHash}', '" . $db->escape_string($action) . "', '{$payloadJson}', {$actorSql}, '{$expiresAt}')"
        );

        return [
            'id' => $id,
            'selection_hash' => $selectionHash,
            'expires_at' => $expiresAt,
            'action' => $action,
            'items' => $items,
            'requires_confirmation' => $requiresConfirmation,
            'confirmation_phrase' => $confirmationPhrase,
        ];
    }

    public function applyDecision(array $input, ?int $actorId = null, array $allowedHallIds = []): array
    {
        global $db;
        xlvask_usage_logs_schema_bootstrap::ensureTables();
        $previewId = trim((string)($input['preview_id'] ?? ''));
        $selectionHash = trim((string)($input['selection_hash'] ?? ''));
        if (!preg_match('/^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i', $previewId)
            || !preg_match('/^[0-9a-f]{64}$/i', $selectionHash)) {
            throw new Exception('Invalid XL Vask decision preview identifiers.');
        }
        $connection = $db->conn();
        $allowedHallIds = $this->normalizeHallIds($allowedHallIds);
        if ($allowedHallIds === []) {
            throw new Exception('No XL Vask hall scope is available for this user.');
        }
        $connection->begin_transaction();
        try {
            $result = $db->query(
                "SELECT * FROM xlvask_automation_decision_previews WHERE id = '" . $db->escape_string($previewId) . "' FOR UPDATE"
            );
            if ($result === false || $result->num_rows < 1) {
                throw new Exception('XL Vask decision preview not found.');
            }
            $preview = $db->fetch_assoc($result);
            if ((int)($preview['created_by'] ?? 0) !== (int)($actorId ?? 0)) {
                throw new Exception('XL Vask decision preview belongs to another user.');
            }
            if (!hash_equals((string)$preview['selection_hash'], $selectionHash)) {
                throw new Exception('XL Vask decision preview changed.');
            }
            if ($preview['applied_at'] !== null || strtotime((string)$preview['expires_at']) < time()) {
                throw new Exception('XL Vask decision preview has expired or was already applied.');
            }
            $payload = json_decode((string)$preview['payload_json'], true);
            if (!is_array($payload)) {
                throw new Exception('XL Vask decision preview is invalid.');
            }
            $requiresConfirmation = count($payload['items'] ?? []) > 1 || in_array($payload['action'] ?? '', ['deny', 'ignore'], true);
            $confirmationPhrase = (string)($payload['confirmation_phrase'] ?? '');
            if ($requiresConfirmation && ($confirmationPhrase === ''
                || !hash_equals($confirmationPhrase, trim((string)($input['confirmation_text'] ?? ''))))) {
                throw new Exception('Confirmation text does not match the preview-issued phrase.');
            }
            $automation = new xlvask_automation_service();
            $results = [];
            foreach ($payload['items'] as $item) {
                $usageId = (int)$item['usage_log_id'];
                $expected = (int)$item['before']['expected_version'];
                $locked = $db->query(
                    "SELECT expected_version, source_hash, HallId FROM xlvask_usage_logs WHERE id = {$usageId} FOR UPDATE"
                );
                $current = $locked !== false && $locked->num_rows > 0 ? $db->fetch_assoc($locked) : null;
                if ($current === null
                    || !in_array(trim((string)$current['HallId']), $allowedHallIds, true)
                    || !self::previewSnapshotMatches($expected, (string)$item['before']['source_hash'], $current)) {
                    throw new Exception("XL Vask usage log {$usageId} changed after preview.");
                }
                $action = (string)$payload['action'];
                if ($action === 'ignore') {
                    $reason = $db->escape_string((string)$payload['reason']);
                    $actorSql = $actorId === null ? 'NULL' : (string)$actorId;
                    if ($db->query(
                        "UPDATE xlvask_usage_logs SET ignored_at = NOW(), ignored_by = {$actorSql}, ignored_reason = '{$reason}',
                         resolution_state = 'ignored', certainty = 'none', planned_action = 'none', expected_version = expected_version + 1
                         WHERE id = {$usageId}"
                    ) === false || $db->conn()->affected_rows !== 1) {
                        throw new Exception('The XL Vask ignore decision could not be applied atomically.');
                    }
                    $results[] = ['usage_log_id' => $usageId, 'resolution_state' => 'ignored'];
                } else {
                    $results[] = $automation->applyBoundDecisionWithinTransaction(
                        $usageId,
                        $action,
                        isset($item['suggestion_id']) ? (int)$item['suggestion_id'] : null,
                        isset($item['candidate_order_id']) && $item['candidate_order_id'] !== null
                            ? (int)$item['candidate_order_id'] : null,
                        $expected,
                        (string)$item['before']['source_hash'],
                        $actorId,
                        $payload['reason'] ?: null
                    );
                }
            }
            if ($db->query(
                "UPDATE xlvask_automation_decision_previews SET applied_at = NOW() WHERE id = '" . $db->escape_string($previewId) . "'"
            ) === false || $db->conn()->affected_rows !== 1) {
                throw new Exception('The XL Vask decision preview could not be finalized atomically.');
            }
            $connection->commit();
        } catch (Throwable $throwable) {
            $connection->rollback();
            throw $throwable;
        }

        return ['applied' => count($results), 'results' => $results, 'failed' => []];
    }

    private function persistRunItems(int $runId, array $results): void
    {
        global $db;
        foreach ($results as $result) {
            $usageId = (int)($result['usage_log_id'] ?? 0);
            if ($usageId < 1) {
                continue;
            }
            $rowResult = $db->query("SELECT * FROM xlvask_usage_logs WHERE id = {$usageId} LIMIT 1");
            if ($rowResult === false || $rowResult->num_rows < 1) {
                continue;
            }
            $row = $db->fetch_assoc($rowResult);
            $durableResult = array_intersect_key($result, array_flip([
                'usage_log_id', 'id', 'status', 'action', 'certainty', 'calibrated_probability',
                'source', 'matched_order_id', 'created_order_id', 'risk_flags', 'policy_version',
            ]));
            $resultJson = $db->escape_string(json_encode($durableResult, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) ?: '{}');
            $db->query(
                "INSERT INTO xlvask_autopilot_run_items
                    (run_id, usage_log_id, wash_id, import_state, resolution_state, certainty, planned_action, source_hash, expected_version, result_json)
                 VALUES ({$runId}, {$usageId}, '" . $db->escape_string((string)$row['WashId']) . "',
                    '" . $db->escape_string((string)($row['import_state'] ?? 'unchanged')) . "',
                    '" . $db->escape_string((string)($result['resolution_state'] ?? $row['resolution_state'] ?? 'needs_review')) . "',
                    '" . $db->escape_string((string)($result['certainty'] ?? $row['certainty'] ?? 'none')) . "',
                    '" . $db->escape_string((string)($result['planned_action'] ?? $row['planned_action'] ?? 'none')) . "',
                    '" . $db->escape_string((string)($row['source_hash'] ?? '')) . "', " . (int)($row['expected_version'] ?? 1) . ", '{$resultJson}')
                 ON DUPLICATE KEY UPDATE resolution_state = VALUES(resolution_state), certainty = VALUES(certainty),
                    planned_action = VALUES(planned_action), result_json = VALUES(result_json), updated_at = NOW()"
            );
        }
    }

    private function formatRun(array $row): array
    {
        $summary = json_decode((string)($row['summary_json'] ?? ''), true);
        return [
            'id' => (int)$row['id'],
            'status' => (string)$row['status'],
            'mode' => (string)$row['mode'],
            'date_from' => $row['date_from'] ?: null,
            'date_to' => $row['date_to'] ?: null,
            'phase' => (string)$row['phase'],
            'processed' => (int)$row['processed'],
            'total' => (int)$row['total'],
            'summary' => is_array($summary) ? $summary : null,
            'warning' => $row['warning'] ?: null,
            'error' => $row['error'] ?: null,
            'created_at' => $row['created_at'],
            'started_at' => $row['started_at'] ?: null,
            'finished_at' => $row['finished_at'] ?: null,
            'attempt_count' => (int)($row['attempt_count'] ?? 0),
            'max_attempts' => (int)($row['max_attempts'] ?? 3),
            'next_attempt_at' => $row['next_attempt_at'] ?: null,
        ];
    }

    public function pruneExpiredData(): array
    {
        global $db;
        xlvask_usage_logs_schema_bootstrap::ensureTables();
        $deleted = [];
        foreach ([
            'previews' => 'DELETE FROM xlvask_automation_decision_previews WHERE expires_at < DATE_SUB(NOW(), INTERVAL 1 DAY)',
            'run_items' => 'DELETE FROM xlvask_autopilot_run_items WHERE created_at < DATE_SUB(NOW(), INTERVAL 90 DAY)',
            'audit_events' => 'DELETE FROM xlvask_automation_audit WHERE created_at < DATE_SUB(NOW(), INTERVAL 730 DAY)',
        ] as $key => $sql) {
            if ($db->query($sql) === false) {
                throw new Exception('XL Vask autopilot retention cleanup failed.');
            }
            $deleted[$key] = (int)$db->conn()->affected_rows;
        }
        return $deleted;
    }

    private function normalizeDate(mixed $value, string $name): ?string
    {
        $value = trim((string)($value ?? ''));
        if ($value === '') {
            return null;
        }
        if (!preg_match('/^\d{4}-\d{2}-\d{2}$/', $value)) {
            throw new Exception("Invalid {$name}.");
        }
        [$year, $month, $day] = array_map('intval', explode('-', $value));
        if (!checkdate($month, $day, $year)) {
            throw new Exception("Invalid {$name}.");
        }
        return $value;
    }

    private function wilsonLowerBound(int $successes, int $total): float
    {
        if ($total < 1) {
            return 0.0;
        }
        $z = 1.959963984540054;
        $p = $successes / $total;
        $zSquared = $z * $z;
        $denominator = 1 + ($zSquared / $total);
        $centre = $p + ($zSquared / (2 * $total));
        $margin = $z * sqrt((($p * (1 - $p)) + ($zSquared / (4 * $total))) / $total);
        return max(0.0, ($centre - $margin) / $denominator);
    }

    private function normalizeHallIds(array $hallIds): array
    {
        return self::normalizeHallScope($hallIds);
    }

    private function quotedHallIds(array $hallIds): string
    {
        global $db;
        return implode(',', array_map(
            static fn(string $id): string => "'" . $db->escape_string($id) . "'",
            $hallIds
        ));
    }

    private function uuidV4(): string
    {
        $data = random_bytes(16);
        $data[6] = chr((ord($data[6]) & 0x0f) | 0x40);
        $data[8] = chr((ord($data[8]) & 0x3f) | 0x80);
        return vsprintf('%s%s-%s-%s-%s-%s%s%s', str_split(bin2hex($data), 4));
    }
}