<?php

declare(strict_types=1);

namespace app\adminapi\service;

use app\common\service\DifyChatService;

/**
 * Shared between FPM workers on one host. Contains no credentials or clinical text.
 * The caller MUST authorize the diagnosis before constructing or reading a control.
 */
final class AssistantRequestControl
{
    private const STOP_LEASE_SECONDS = 1.25;
    private const STOP_RETRY_DELAY_SECONDS = 0.25;
    private const STOP_RETENTION_SECONDS = 30;
    private const MAX_STOP_ATTEMPTS = 2;
    private const VERIFICATION_BUDGET_SECONDS = 12.0;
    private const LEGACY_VERIFICATION_BUDGET_SECONDS = 2.0;
    private const VERIFICATION_IO_SECONDS = 1.0;
    private const VERIFICATION_LEASE_SECONDS = 1.25;
    private const VERIFICATION_RETRY_SECONDS = 0.25;
    private const VERIFICATION_FINAL_IO_RESERVE_SECONDS = 1.0;
    private const MAX_VERIFICATION_ATTEMPTS = 2;

    private string $path;
    private int $ttl;
    private string $directory;

    public function __construct(int $adminId, int $diagnosisId, string $requestId, ?string $directory = null, int $ttl = 1200)
    {
        if ($adminId <= 0 || $diagnosisId <= 0 || !self::validRequestId($requestId)) {
            throw new \InvalidArgumentException('Invalid assistant request identity');
        }
        $directory = $directory ?? app()->getRuntimePath() . 'assistant-requests';
        if (!is_dir($directory) && !@mkdir($directory, 0700, true) && !is_dir($directory)) {
            throw new \RuntimeException('Assistant request control unavailable');
        }
        $this->path = rtrim($directory, '/\\') . DIRECTORY_SEPARATOR
            . hash('sha256', $adminId . ':' . $diagnosisId . ':' . strtolower($requestId)) . '.json';
        $this->ttl = max(600, $ttl);
        $this->directory = $directory;
        $this->collectExpired();
    }

    public static function validRequestId(string $id): bool
    {
        return preg_match('/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i', $id) === 1;
    }

    /** A cancel tombstone created before prepare is retained; a replay never restarts work. */
    public function begin(): bool
    {
        return $this->update(static function (array &$state): bool {
            if (!empty($state['begun']) || self::hasUnconfirmedIdentity($state)) {
                return false;
            }
            $state['begun'] = true;
            $state['status'] = !empty($state['cancel_requested']) ? 'cancel_requested' : 'running';
            return true;
        });
    }

    public function shouldAbort(): bool
    {
        return $this->update(static fn (array &$state): bool => !empty($state['cancel_requested']), false);
    }

    /** Read-only gate before every new generation/compatibility attempt. */
    public function canStartAttempt(): bool
    {
        return $this->update(static fn (array &$state): bool => self::canStartState($state), false);
    }

    private static function canStartState(array $state, string $ignoreSubmission = ''): bool
    {
        if (empty($state['begun']) || !empty($state['cancel_requested']) || !empty($state['timeout_reached'])
            || in_array($state['status'] ?? '', ['done', 'error', 'cancelled'], true)) return false;
        if (self::submissionCounts($state, $ignoreSubmission)['pending'] > 0) return false;
        if (is_array($state['upstream'] ?? null) && !self::independentlyTerminal($state)) return false;
        foreach ($state['handoff_cleanup'] ?? [] as $cleanup) {
            if (is_array($cleanup['upstream'] ?? null) && !self::independentlyTerminal($cleanup)) return false;
        }
        return true;
    }

    /** Atomic ownership record immediately before this generation's first POST execution. */
    public function beginSubmission(string $profile, string $user): string
    {
        if (!in_array($profile, ['qwen', 'openai'], true) || strlen($user) > 128) throw new \RuntimeException('Invalid submission scope');
        $token = bin2hex(random_bytes(12));
        return $this->update(static function (array &$state) use ($profile, $user, $token): string {
            if (!self::canStartState($state)) throw new \RuntimeException('UPSTREAM_CLEANUP_PENDING');
            self::releaseTask($state);
            $state['active_submission_id'] = $token;
            $state['upstream_stop_confirmed'] = false;
            $state['upstream_terminal_confirmed'] = false;
            $state['upstream_terminal_status'] = '';
            foreach (['producer_terminal_status', 'producer_terminal_observed_at', 'persisted_run_status',
                'persisted_run_observed_at', 'persisted_terminal_confirmed', 'lifecycle_conflict', 'run_status_observations',
                'transfer_closed', 'transfer_closed_at', 'generation_verification_elapsed_ms', 'stop_accepted_at',
                'stop_requested', 'stop_retry_until'] as $field) unset($state[$field]);
            $state['submission_sequence'] = (int) ($state['submission_sequence'] ?? 0) + 1;
            $state['submitted_attempts'][$token] = ['generation' => $state['submission_sequence'], 'profile' => $profile,
                'user' => $user, 'submitted_at' => microtime(true), 'transfer_closed' => false,
                'task_id' => '', 'workflow_run_id' => '', 'disposition' => 'unconfirmed'];
            return $token;
        });
    }

    /** Empty-task closure is authoritative only for its own admitted attempt token. */
    public function closeSubmittedGeneration(string $token, string $taskId, string $disposition, callable $lookup): array
    {
        if (!in_array($disposition, ['unconfirmed', 'not_submitted', 'rejected_before_execution'], true)) {
            throw new \InvalidArgumentException('Invalid submission disposition');
        }
        $bound = $this->update(static function (array &$state) use ($token, $taskId, $disposition): ?string {
            if (!preg_match('/\A[0-9a-f]{24}\z/D', $token) || !isset($state['submitted_attempts'][$token])) return null;
            $record =& $state['submitted_attempts'][$token];
            if ($record['task_id'] !== $taskId) return null;
            if (empty($record['transfer_closed'])) {
                $record['transfer_closed'] = true;
                $record['transfer_closed_at'] = microtime(true);
                if ($taskId === '') $record['disposition'] = $disposition;
                if (($state['active_submission_id'] ?? '') === $token && $taskId === '') {
                    $state['transfer_closed'] = true;
                    $state['transfer_closed_at'] = $record['transfer_closed_at'];
                }
            }
            return $record['task_id'];
        });
        if ($bound === null) return ['confirmed' => false, 'pending' => true, 'reason' => 'stale_submission'];
        if ($bound === '') {
            $pending = $this->update(static fn (array &$state): bool =>
                !in_array($state['submitted_attempts'][$token]['disposition'], ['not_submitted', 'rejected_before_execution'], true), false);
            return ['confirmed' => false, 'pending' => $pending,
                'reason' => $pending ? 'identity_unavailable' : 'pre_execution_closed'];
        }
        // Exact owned task/run validation, lookup claims and budgets stay in the existing path.
        return $this->closeGeneration($bound, $lookup);
    }

    /** Submission resolution never borrows a same-ID record from another attempt. */
    private static function submissionCounts(array $state, string $ignore = ''): array
    {
        $counts = ['total' => 0, 'closed' => 0, 'pending' => 0, 'unidentified' => 0, 'additional_pending' => 0, 'pre_execution' => 0, 'not_submitted' => 0, 'rejected_before_execution' => 0];
        foreach ($state['submitted_attempts'] ?? [] as $token => $submission) {
            ++$counts['total'];
            if (!empty($submission['transfer_closed'])) ++$counts['closed'];
            if ($token === $ignore) continue;
            if (!empty($submission['transfer_closed']) && ($submission['task_id'] ?? '') === ''
                && in_array($submission['disposition'] ?? '', ['not_submitted', 'rejected_before_execution'], true)) {
                ++$counts['pre_execution'];
                ++$counts[$submission['disposition']]; continue;
            }
            $resolved = false; $activePending = false;
            $targets = array_merge([$state], array_values($state['handoff_cleanup'] ?? []));
            foreach ($targets as $target) {
                $records = array_values($target['completed_generations'] ?? []);
                if (is_array($target['upstream'] ?? null)) $records[] = $target;
                foreach ($records as $record) {
                    $identity = $record['upstream'] ?? [];
                    if (($identity['submission_id'] ?? '') !== $token || ($submission['task_id'] ?? '') === ''
                        || ($identity['task_id'] ?? null) !== $submission['task_id']
                        || ($identity['workflow_run_id'] ?? null) !== ($submission['workflow_run_id'] ?? '')
                        || ($identity['profile'] ?? null) !== ($submission['profile'] ?? null)
                        || ($identity['user'] ?? null) !== ($submission['user'] ?? null)
                        || ($identity['generation'] ?? null) !== ($submission['control_generation'] ?? null)) continue;
                    if (self::independentlyTerminal($record) && !empty($submission['transfer_closed'])) $resolved = true;
                    elseif (($target['upstream']['submission_id'] ?? '') === $token) $activePending = true;
                }
            }
            if (!$resolved) {
                ++$counts['pending'];
                if (($submission['task_id'] ?? '') === '' || ($submission['workflow_run_id'] ?? '') === '') ++$counts['unidentified'];
                if (!$activePending) ++$counts['additional_pending'];
            }
        }
        return $counts;
    }

    /** An already-created conflicting task must be stopped without user-cancel intent. */
    public function hasHandoffTask(string $taskId): bool
    {
        return $this->update(static fn (array &$state): bool =>
            ($state['handoff_cleanup'][$taskId]['upstream']['task_id'] ?? '') === $taskId, false);
    }

    /** Keep normal and exceptional cleanup leases in the same request's private file. */
    private static function &taskState(array &$state, ?string $taskId): array
    {
        if ($taskId === null || ($state['upstream']['task_id'] ?? '') === $taskId) return $state;
        if (isset($state['handoff_cleanup'][$taskId])) return $state['handoff_cleanup'][$taskId];
        $missing = [];
        return $missing;
    }

    public function bindTask(string $profile, string $user, string $taskId, string $workflowRunId = '', string $submissionId = ''): void
    {
        if (!in_array($profile, ['qwen', 'openai'], true)
            || !self::validUpstreamId($taskId) || strlen($user) > 128
            || ($workflowRunId !== '' && !self::validUpstreamId($workflowRunId))) {
            throw new \RuntimeException('Invalid upstream task identity');
        }
        $outcome = $this->update(static function (array &$state) use ($profile, $user, $taskId, $workflowRunId, $submissionId): string {
            if ($submissionId !== '') {
                $submission = $state['submitted_attempts'][$submissionId] ?? null;
                if (!is_array($submission) || !empty($submission['transfer_closed'])
                    || $submission['profile'] !== $profile || $submission['user'] !== $user
                    || ($submission['task_id'] !== '' && $submission['task_id'] !== $taskId)
                    || ($submission['workflow_run_id'] !== '' && $workflowRunId !== '' && $submission['workflow_run_id'] !== $workflowRunId)) return 'UPSTREAM_TASK_IDENTITY_CONFLICT';
            }
            $bind = static function (array &$state) use ($profile, $user, $taskId, $workflowRunId, $submissionId): string {
                $completed = $state['completed_generations'] ?? [];
                foreach ($state['handoff_cleanup'] ?? [] as $cleanup) array_push($completed, ...array_values($cleanup['completed_generations'] ?? []));
                foreach ($completed as $record) {
                    if (($record['upstream']['task_id'] ?? '') === $taskId
                        || ($workflowRunId !== '' && ($record['upstream']['workflow_run_id'] ?? '') === $workflowRunId)) return 'UPSTREAM_TASK_IDENTITY_CONFLICT';
                }
                $previous = $state['upstream'] ?? [];
                if (($previous['task_id'] ?? '') === $taskId) {
                    return self::enrichRun($state, $profile, $user, $workflowRunId)
                        ? '' : 'UPSTREAM_TASK_IDENTITY_CONFLICT';
                }
                if (isset($state['handoff_cleanup'][$taskId])) {
                    return self::enrichRun($state['handoff_cleanup'][$taskId], $profile, $user, $workflowRunId)
                        ? 'UPSTREAM_TASK_HANDOFF_BLOCKED' : 'UPSTREAM_TASK_IDENTITY_CONFLICT';
                }
                // The first identity may arrive after the deadline/cancel frame.
                // It belongs to the already-started transfer and must remain the
                // primary cleanup target, not be mistaken for a new generation.
                $lateFirst = !empty($state['begun']) && empty($previous) && (empty($state['task_generation']) || $submissionId !== '');
                $state['task_generation'] = (int) ($state['task_generation'] ?? 0) + 1;
                $incoming = ['profile' => $profile, 'user' => $user, 'task_id' => $taskId,
                    'workflow_run_id' => $workflowRunId, 'generation' => $state['task_generation']];
                if ($submissionId !== '') $incoming['submission_id'] = $submissionId;
                if (!$lateFirst && !self::canStartState($state, $submissionId)) {
                    // The pre-start gate normally prevents this race. If a task
                    // already exists remotely, retain BOTH identities and its own
                    // bounded stop lease; never replace the unconfirmed old task.
                    $state['handoff_cleanup'][$taskId] ??= ['upstream' => $incoming,
                        'stop_requested' => true, 'stop_attempts' => 0,
                        'stop_retry_until' => microtime(true) + self::STOP_RETENTION_SECONDS];
                    return 'UPSTREAM_TASK_HANDOFF_BLOCKED';
                }
                $state['upstream'] = $incoming;
                $state['upstream_stop_confirmed'] = false;
                $state['upstream_terminal_confirmed'] = false;
                $state['upstream_terminal_status'] = '';
                foreach (['producer_terminal_status', 'producer_terminal_observed_at', 'persisted_run_status',
                    'persisted_run_observed_at', 'persisted_terminal_confirmed', 'lifecycle_conflict', 'run_status_observations',
                    'transfer_closed', 'transfer_closed_at', 'generation_verification_elapsed_ms', 'stop_accepted_at'] as $field) unset($state[$field]);
                $state['stop_attempts'] = 0;
                // These belong to the previous generation. In particular an old
                // expired retention deadline must not disable a later real cancel.
                unset($state['stop_claim'], $state['stop_retry_after']);
                unset($state['verification_claim'], $state['verification_attempts'],
                    $state['verification_started_at'], $state['verification_deadline_at'], $state['verification_retry_after']);
                if (!$lateFirst) unset($state['stop_requested'], $state['stop_retry_until']);
                return '';
            };
            $result = $bind($state);
            if ($submissionId !== '' && in_array($result, ['', 'UPSTREAM_TASK_HANDOFF_BLOCKED'], true)) {
                $target =& self::taskState($state, $taskId);
                if (isset($target['upstream']) && ($target['upstream']['submission_id'] ?? $submissionId) === $submissionId) {
                    $target['upstream']['submission_id'] = $submissionId;
                    $state['submitted_attempts'][$submissionId]['task_id'] = $taskId;
                    $state['submitted_attempts'][$submissionId]['workflow_run_id'] = $target['upstream']['workflow_run_id'];
                    $state['submitted_attempts'][$submissionId]['control_generation'] = $target['upstream']['generation'];
                } else return 'UPSTREAM_TASK_IDENTITY_CONFLICT';
            }
            return $result;
        });
        if ($outcome !== '') throw new \RuntimeException($outcome);
    }

    private static function validUpstreamId(string $id): bool
    {
        return preg_match('/\A[a-zA-Z0-9_-]{1,128}\z/', $id) === 1;
    }

    /** Same-generation enrichment never changes stop, cancel or verification state. */
    private static function enrichRun(array &$target, string $profile, string $user, string $workflowRunId): bool
    {
        $task = $target['upstream'] ?? [];
        if (($task['profile'] ?? '') !== $profile || ($task['user'] ?? '') !== $user) return false;
        $captured = (string) ($task['workflow_run_id'] ?? '');
        if ($workflowRunId !== '' && $captured !== '' && $captured !== $workflowRunId) return false;
        if ($workflowRunId !== '' && $captured === '') $target['upstream']['workflow_run_id'] = $workflowRunId;
        return true;
    }

    public function clearTask(?string $taskId = null): void
    {
        $this->update(static function (array &$state) use ($taskId): void {
            $target =& self::taskState($state, $taskId);
            if ($taskId !== null && ($target['upstream']['task_id'] ?? '') !== $taskId) return;
            if (is_array($target['upstream'] ?? null)) {
                $target['transfer_closed'] = true;
                $target['transfer_closed_at'] ??= microtime(true);
            }
            self::releaseTask($target);
        });
    }

    public function markStop(bool $confirmed, ?string $taskId = null): void
    {
        $this->update(static function (array &$state) use ($confirmed, $taskId): void {
            $target =& self::taskState($state, $taskId);
            if ($taskId !== null && ($target['upstream']['task_id'] ?? '') !== $taskId) return;
            $target['upstream_stop_confirmed'] = !empty($target['upstream_stop_confirmed']) || $confirmed;
            if ($confirmed) $target['stop_accepted_at'] ??= microtime(true);
        });
    }

    /** Bound producer receipt only. It never proves the independent run status. */
    public function markTerminal(string $taskId, string $status, string $workflowRunId = ''): void
    {
        if (!in_array($status, ['stopped', 'succeeded', 'failed'], true)) return;
        $this->update(static function (array &$state) use ($taskId, $status, $workflowRunId): void {
            $target =& self::taskState($state, $taskId);
            if (($target['upstream']['task_id'] ?? '') !== $taskId) return;
            $captured = (string) ($target['upstream']['workflow_run_id'] ?? '');
            // Legacy live-stream task-only receipts remain compatible. They do
            // not establish a run binding and cannot enable independent lookup.
            if ($captured !== '' ? $captured !== $workflowRunId : $workflowRunId !== '') return;
            if (isset($target['producer_terminal_status'])) {
                if ($target['producer_terminal_status'] !== $status) $target['lifecycle_conflict'] = true;
                return;
            }
            $target['producer_terminal_status'] = $status;
            $target['producer_terminal_observed_at'] = microtime(true);
            if (isset($target['persisted_run_status']) && $target['persisted_run_status'] !== $status) $target['lifecycle_conflict'] = true;
        });
    }

    /** Called only after this generation's curl handle has closed, never from its callbacks. */
    public function closeGeneration(string $taskId, callable $lookup): array
    {
        if (!self::validUpstreamId($taskId)) return ['confirmed' => false, 'pending' => true, 'reason' => 'identity_unavailable'];
        $this->clearTask($taskId);
        return $this->reconcileTerminalIdentity($taskId, $lookup);
    }

    private static function independentlyTerminal(array $target): bool
    {
        return ($target['persisted_terminal_confirmed'] ?? null) === true
            && !empty($target['upstream_terminal_confirmed'])
            && in_array($target['persisted_run_status'] ?? null, ['stopped', 'succeeded', 'failed'], true);
    }

    private static function observationTime($value): ?float
    {
        return (is_float($value) || is_int($value)) && is_finite((float) $value) && $value >= 0 ? (float) $value : null;
    }

    /**
     * Separately invoked recovery for an already closed transport. No built-in
     * lookup, network call or public route exists. The trusted read-only adapter
     * receives only the exact private identity and an I/O timeout in milliseconds;
     * it MUST honor that timeout. Late/throwing/malformed results fail closed.
     * Arbitrary PHP callbacks cannot be preempted; elapsed time is checked again
     * before applying any result. Neither returned bodies nor error prose persist.
     * @return array{confirmed:bool,pending:bool,reason:string}
     */
    public function reconcileTerminal(string $expectedTaskId, callable $lookup): array
    {
        return $this->reconcileTerminalIdentity($expectedTaskId, $lookup);
    }

    /**
     * Recover only this authorized request's original captured pairs after finish.
     * The shared monotonic budget survives repeated calls and covers all targets,
     * I/O and retry spacing. The adapter must honor each supplied I/O timeout;
     * arbitrary PHP callbacks cannot be preempted and late results are rejected.
     * No private identity, response body or exception metadata leaves this method.
     * @return array<string,bool|int|string>
     */
    public function reconcileClosedTasks(callable $lookup): array
    {
        $invokedAt = hrtime(true) / 1e9;
        $snapshot = $this->update(static function (array &$state) use ($invokedAt): array {
            $closed = !empty($state['transport_closed']);
            $tasks = [];
            foreach (array_merge([$state], array_values($state['handoff_cleanup'] ?? [])) as $target) {
                if (!is_array($target) || !is_array($target['upstream'] ?? null) || self::independentlyTerminal($target)) continue;
                $identity = self::verificationIdentity($target);
                if ($closed && $identity !== null) $tasks[] = $identity;
            }
            $deadline = $invokedAt;
            if ($tasks !== []) {
                $existingBudget = isset($state['closed_verification_started_at']);
                $started = (float) ($state['closed_verification_started_at'] ?? $invokedAt);
                $state['closed_verification_started_at'] = $started;
                // Freeze new windows; existing start-only controls retain their
                // original two-second deadline rather than gaining a new budget.
                $originalDeadline = min($started + self::VERIFICATION_BUDGET_SECONDS,
                    (float) ($state['closed_verification_deadline_at'] ?? ($started + ($existingBudget
                        ? self::LEGACY_VERIFICATION_BUDGET_SECONDS : self::VERIFICATION_BUDGET_SECONDS))));
                $state['closed_verification_deadline_at'] = $originalDeadline;
                // A host monotonic-clock reset must not renew a persisted budget.
                $deadline = $started > $invokedAt || !is_finite($originalDeadline) ? $invokedAt : $originalDeadline;
            }
            return ['closed' => $closed, 'tasks' => $tasks, 'deadline' => $deadline];
        });
        $attempts = 0;
        $queried = 0;
        $confirmed = 0;
        $exhausted = false;
        foreach ($snapshot['tasks'] as $identity) {
            if (hrtime(true) / 1e9 >= $snapshot['deadline']) {
                $exhausted = true;
                break;
            }
            $targetAttempts = 0;
            $boundedLookup = static function (array $task, int $timeoutMs) use ($lookup, &$attempts, &$targetAttempts) {
                ++$attempts;
                ++$targetAttempts;
                return $lookup($task, $timeoutMs);
            };
            $result = $this->reconcileTerminalIdentity($identity['task_id'], $boundedLookup, $snapshot['deadline'], $identity);
            if ($targetAttempts > 0) ++$queried;
            if ($result['confirmed']) ++$confirmed;
            if ($result['reason'] === 'budget_exhausted' && hrtime(true) / 1e9 >= $snapshot['deadline']) $exhausted = true;
        }
        $pending = $this->update(static function (array &$state): int {
            $count = 0;
            foreach (array_merge([$state], array_values($state['handoff_cleanup'] ?? [])) as $target) {
                if (is_array($target) && is_array($target['upstream'] ?? null) && !self::independentlyTerminal($target)) ++$count;
            }
            return $count + self::submissionCounts($state)['additional_pending'];
        }, false);
        return [
            'status' => !$snapshot['closed'] ? 'transport_open' : ($pending === 0 ? 'complete' : ($exhausted ? 'budget_exhausted' : 'pending')),
            'transport_closed' => $snapshot['closed'],
            'eligible_task_count' => count($snapshot['tasks']),
            'queried_task_count' => $queried,
            'lookup_attempt_count' => $attempts,
            'confirmed_task_count' => $confirmed,
            'upstream_pending_task_count' => $pending,
            'budget_exhausted' => $exhausted,
            'elapsed_ms' => max(0, (int) floor((hrtime(true) / 1e9 - $invokedAt) * 1000)),
        ];
    }

    /** The optional caller deadline only narrows each generation's persisted budget. */
    private function reconcileTerminalIdentity(string $expectedTaskId, callable $lookup, ?float $invocationDeadline = null,
        ?array $expectedIdentity = null): array
    {
        while (true) {
            $token = bin2hex(random_bytes(12));
            $claim = $this->update(static function (array &$state) use ($expectedTaskId, $token, $invocationDeadline, $expectedIdentity): array {
                $target =& self::taskState($state, $expectedTaskId);
                $task = self::verificationIdentity($target);
                if (!is_array($target['upstream'] ?? null)) {
                    return ['result' => ['confirmed' => false, 'pending' => false, 'reason' => 'task_unavailable']];
                }
                if (self::independentlyTerminal($target)) {
                    return ['result' => ['confirmed' => true, 'pending' => false, 'reason' => 'already_confirmed']];
                }
                if ($task === null || $task['task_id'] !== $expectedTaskId) {
                    return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'identity_unavailable']];
                }
                if ($expectedIdentity !== null && $task !== $expectedIdentity) {
                    return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'stale_result']];
                }
                if (empty($state['transport_closed']) && empty($target['transfer_closed'])) {
                    return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'transport_open']];
                }
                $now = hrtime(true) / 1e9;
                $existingBudget = isset($target['verification_started_at']);
                $started = (float) ($target['verification_started_at'] ?? $now);
                $originalDeadline = min($started + self::VERIFICATION_BUDGET_SECONDS,
                    (float) ($target['verification_deadline_at'] ?? ($started + ($existingBudget
                        ? self::LEGACY_VERIFICATION_BUDGET_SECONDS : self::VERIFICATION_BUDGET_SECONDS))));
                // A monotonic clock reset (e.g. host reboot) never renews a budget.
                $deadline = min($originalDeadline, $invocationDeadline ?? INF);
                $remaining = $deadline - $now;
                if ($started > $now || !is_finite($deadline) || $remaining <= 0) {
                    return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'budget_exhausted']];
                }
                $attempts = (int) ($target['verification_attempts'] ?? 0);
                if ($attempts >= self::MAX_VERIFICATION_ATTEMPTS) {
                    return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'attempts_exhausted']];
                }
                if ((float) ($target['verification_claim']['until'] ?? 0) > $now) {
                    return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'verification_in_progress']];
                }
                $retryAt = (float) ($target['verification_retry_after'] ?? 0);
                if ($attempts === self::MAX_VERIFICATION_ATTEMPTS - 1) {
                    // Keep the first observation fast. Spend the final finite read
                    // after the owned pool's bounded ten-second teardown, reserving
                    // one second for I/O within the SAME persisted window.
                    $retryAt = max($retryAt, $deadline - self::VERIFICATION_FINAL_IO_RESERVE_SECONDS);
                }
                $wait = $retryAt - $now;
                if ($wait > 0) return ['wait_us' => (int) ceil(min($wait, $remaining) * 1e6)];
                $timeoutMs = (int) floor(min(self::VERIFICATION_IO_SECONDS, $remaining) * 1000);
                if ($timeoutMs < 1) {
                    return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'budget_exhausted']];
                }
                $target['verification_started_at'] = $started;
                $target['verification_deadline_at'] = $originalDeadline;
                $target['verification_attempts'] = $attempts + 1;
                $target['verification_claim'] = ['token' => $token, 'until' => $now + self::VERIFICATION_LEASE_SECONDS];
                return ['task' => $task, 'timeout_ms' => $timeoutMs,
                    'submission_owner' => $target['upstream']['submission_id'] ?? '', 'deadline' => $deadline];
            });
            if (isset($claim['result'])) return $claim['result'];
            if (isset($claim['wait_us'])) {
                usleep($claim['wait_us']);
                continue;
            }
            $task = $claim['task'];
            $status = null;
            $observedStatus = null;
            $reason = 'lookup_failed';
            $ioStarted = hrtime(true) / 1e9;
            // No request-file or GC lock is held across the injected I/O.
            try {
                $response = $lookup($task, $claim['timeout_ms']);
                $observedStatus = self::observedStatus($response, $task);
                $status = in_array($observedStatus, ['stopped', 'succeeded', 'failed'], true) ? $observedStatus : null;
                $reason = $status === null ? 'unconfirmed' : 'confirmed';
            } catch (\Throwable $ignored) {}
            $ioFinished = hrtime(true) / 1e9;
            if ($ioFinished - $ioStarted > $claim['timeout_ms'] / 1000 || $ioFinished >= $claim['deadline']) {
                $status = null;
                $reason = 'budget_exhausted';
            }
            $deadline = $claim['deadline'];
            $submissionOwner = $claim['submission_owner'];
            $result = $this->update(static function (array &$state) use ($task, $token, $status, $observedStatus, $reason, $deadline, $submissionOwner): array {
                $target =& self::taskState($state, $task['task_id']);
                if (self::verificationIdentity($target) !== $task
                    || ($target['upstream']['submission_id'] ?? '') !== $submissionOwner
                    || ($target['verification_claim']['token'] ?? '') !== $token) {
                    return ['confirmed' => false, 'pending' => true, 'reason' => 'stale_result'];
                }
                unset($target['verification_claim']);
                $target['verification_retry_after'] = hrtime(true) / 1e9 + self::VERIFICATION_RETRY_SECONDS;
                $target['generation_verification_elapsed_ms'] = max(0, (int) floor(
                    (hrtime(true) / 1e9 - $target['verification_started_at']) * 1000));
                if ($observedStatus !== null) {
                    $at = microtime(true);
                    $target['persisted_run_status'] = $observedStatus;
                    $target['persisted_run_observed_at'] = $at;
                    $target['run_status_observations'][] = ['status' => $observedStatus, 'observed_at' => $at,
                        'within_budget' => $reason !== 'budget_exhausted'];
                    if (isset($target['producer_terminal_status']) && $target['producer_terminal_status'] !== $observedStatus) $target['lifecycle_conflict'] = true;
                }
                if (self::independentlyTerminal($target)) {
                    return ['confirmed' => true, 'pending' => false, 'reason' => 'already_confirmed'];
                }
                if ($status !== null && hrtime(true) / 1e9 < $deadline) {
                    $target['upstream_terminal_confirmed'] = true;
                    $target['upstream_terminal_status'] = $status;
                    $target['persisted_terminal_confirmed'] = true;
                    self::releaseTask($target);
                    return ['confirmed' => true, 'pending' => false, 'reason' => 'confirmed'];
                }
                return ['confirmed' => false, 'pending' => true,
                    'reason' => hrtime(true) / 1e9 >= $deadline ? 'budget_exhausted' : $reason];
            });
            if ($result['confirmed'] || $result['reason'] === 'stale_result' || $reason === 'budget_exhausted') return $result;
        }
    }

    /** Only original captured pairs with a local generation are lookup-capable. */
    private static function verificationIdentity(array $target): ?array
    {
        $task = $target['upstream'] ?? [];
        if (!is_array($task) || !is_string($task['task_id'] ?? null) || !self::validUpstreamId($task['task_id'])
            || !is_string($task['workflow_run_id'] ?? null) || !self::validUpstreamId($task['workflow_run_id'])
            || !in_array($task['profile'] ?? '', ['qwen', 'openai'], true)
            || !is_string($task['user'] ?? null) || strlen($task['user']) > 128
            || !is_int($task['generation'] ?? null) || $task['generation'] <= 0) return null;
        return ['profile' => $task['profile'], 'user' => $task['user'], 'task_id' => $task['task_id'],
            'workflow_run_id' => $task['workflow_run_id'], 'generation' => $task['generation']];
    }

    private static function observedStatus($response, array $task): ?string
    {
        if (!is_array($response) || ($response['ok'] ?? null) !== true
            || ($response['workflow_run_id'] ?? null) !== $task['workflow_run_id']
            || !in_array($response['status'] ?? null, ['scheduled', 'running', 'succeeded', 'failed', 'stopped', 'partial-succeeded', 'paused'], true)) return null;
        foreach (['task_id', 'user', 'profile'] as $field) {
            if (array_key_exists($field, $response) && $response[$field] !== $task[$field]) return null;
        }
        return $response['status'];
    }

    /** Safe inside a transport callback: writes intent and never performs network I/O. */
    public function signalCancel(): void
    {
        $this->update(static function (array &$state): void {
            if (in_array($state['status'] ?? '', ['done', 'error'], true)) return;
            $state['cancel_requested'] = true;
            if (($state['status'] ?? '') !== 'cancelled') $state['status'] = 'cancel_requested';
            $state['stop_retry_until'] = $state['stop_retry_until'] ?? microtime(true) + self::STOP_RETENTION_SECONDS;
        });
    }

    /** A generation deadline requests remote cleanup without becoming a user cancellation. */
    public function signalTimeout(): void
    {
        $this->update(static function (array &$state): void {
            $state['timeout_reached'] = true;
            $state['stop_requested'] = true;
            $state['stop_retry_until'] = $state['stop_retry_until'] ?? microtime(true) + self::STOP_RETENTION_SECONDS;
        });
    }

    /** Rejecting clinical output stops work without pretending the doctor cancelled. */
    public function signalValidationFailure(): void
    {
        $this->update(static function (array &$state): void {
            $state['validation_rejected'] = true;
            $state['stop_requested'] = true;
            $state['stop_retry_until'] = $state['stop_retry_until'] ?? microtime(true) + self::STOP_RETENTION_SECONDS;
        });
    }

    /** Safe client view. A requested cancellation is never reported as confirmed stop. */
    public function requestCancel(?callable $stop = null): array
    {
        $this->signalCancel();
        $this->stopBoundTask($stop);
        $cleanupIds = $this->update(static fn (array &$state): array => array_keys($state['handoff_cleanup'] ?? []), false);
        foreach ($cleanupIds as $taskId) $this->stopBoundTask($stop, (string) $taskId);
        return $this->publicState();
    }

    /**
     * Shared by the cancel endpoint and stream owner. The lease exceeds the 1s stop
     * transport budget; a crashed claimant can be replaced, while accepted stops
     * never repeat. Failed tasks remain eligible for one short-lived compensation.
     * @return array{accepted:bool,pending:bool}
     */
    public function stopBoundTask(?callable $stop = null, ?string $expectedTaskId = null): array
    {
        $token = bin2hex(random_bytes(12));
        $claim = $this->update(static function (array &$state) use ($token, $expectedTaskId): array {
            $target =& self::taskState($state, $expectedTaskId);
            $now = microtime(true);
            $task = $target['upstream'] ?? null;
            $accepted = !empty($target['upstream_stop_confirmed']);
            if (!is_array($task) || ($expectedTaskId !== null && $task['task_id'] !== $expectedTaskId)) {
                return ['accepted' => false, 'pending' => false];
            }
            // A cancel worker may load a task written by the previous local
            // release. Upgrade its private identity atomically before claiming.
            if (!isset($task['generation'])) {
                $target['task_generation'] = (int) ($target['task_generation'] ?? 0) + 1;
                $task['generation'] = $target['task_generation'];
                $target['upstream'] = $task;
            }
            if ($accepted || self::independentlyTerminal($target) || isset($target['producer_terminal_status'])
                || (empty($target['cancel_requested']) && empty($target['stop_requested']))
                || (float) ($target['stop_retry_until'] ?? 0) <= $now) {
                return ['accepted' => $accepted, 'pending' => false];
            }
            if ((float) ($target['stop_claim']['until'] ?? 0) > $now
                || (float) ($target['stop_retry_after'] ?? 0) > $now) {
                return ['accepted' => false, 'pending' => true];
            }
            if ((int) ($target['stop_attempts'] ?? 0) >= self::MAX_STOP_ATTEMPTS) {
                return ['accepted' => false, 'pending' => false];
            }
            $target['stop_claim'] = ['token' => $token, 'until' => $now + self::STOP_LEASE_SECONDS];
            $target['stop_attempts'] = (int) ($target['stop_attempts'] ?? 0) + 1;
            return ['task' => $task, 'accepted' => false, 'pending' => true];
        });
        if (!isset($claim['task'])) return $claim;
        $task = $claim['task'];
        $confirmed = false;
        // Never hold either file lock while waiting for the remote stop endpoint.
        try {
            $confirmed = ($stop !== null ? $stop($task) : DifyChatService::stopTask(
                $task['profile'], $task['task_id'], $task['user']
            )) === true;
        } catch (\Throwable $ignored) {}
        return $this->update(static function (array &$state) use ($task, $token, $confirmed): array {
            $target =& self::taskState($state, $task['task_id']);
            if (($target['upstream']['generation'] ?? 0) !== $task['generation']
                || ($target['upstream']['task_id'] ?? '') !== $task['task_id']
                || ($target['upstream']['submission_id'] ?? '') !== ($task['submission_id'] ?? '')
                || ($target['stop_claim']['token'] ?? '') !== $token) {
                return ['accepted' => false, 'pending' => false];
            }
            $target['upstream_stop_confirmed'] = !empty($target['upstream_stop_confirmed']) || $confirmed;
            if ($confirmed) $target['stop_accepted_at'] ??= microtime(true);
            unset($target['stop_claim']);
            $target['stop_retry_after'] = microtime(true) + self::STOP_RETRY_DELAY_SECONDS;
            return ['accepted' => !empty($target['upstream_stop_confirmed']), 'pending' => false];
        });
    }

    private static function releaseTask(array &$state): void
    {
        // The stop I/O retry window may expire, but an unconfirmed identity
        // survives TTL reset/GC. Expiry is not a terminal.
        if (is_array($state['upstream'] ?? null) && !self::independentlyTerminal($state)) {
            return;
        }
        if (is_array($state['upstream'] ?? null)) {
            $fields = ['upstream', 'upstream_stop_confirmed', 'stop_accepted_at', 'producer_terminal_status',
                'producer_terminal_observed_at', 'persisted_run_status', 'persisted_run_observed_at',
                'persisted_terminal_confirmed', 'upstream_terminal_confirmed', 'upstream_terminal_status',
                'lifecycle_conflict', 'run_status_observations', 'transfer_closed', 'transfer_closed_at',
                'verification_attempts', 'verification_started_at', 'verification_deadline_at', 'generation_verification_elapsed_ms'];
            $state['completed_generations'][$state['upstream']['generation']] = array_intersect_key($state, array_flip($fields));
        }
        unset($state['upstream'], $state['stop_claim']);
    }

    public function finish(string $status): void
    {
        if (!in_array($status, ['done', 'error', 'cancelled'], true)) {
            throw new \InvalidArgumentException('Invalid assistant terminal state');
        }
        $this->update(static function (array &$state) use ($status): void {
            if (in_array($state['status'] ?? '', ['done', 'error', 'cancelled'], true)) {
                return;
            }
            $state['status'] = !empty($state['cancel_requested']) ? 'cancelled' : $status;
            $state['transport_closed'] = true;
            self::releaseTask($state);
        });
    }

    public function publicState(): array
    {
        return $this->update(static function (array &$state): array {
            $pending = is_array($state['upstream'] ?? null) && !self::independentlyTerminal($state) ? 1 : 0;
            $stopConfirmed = !empty($state['upstream_stop_confirmed']);
            $terminalConfirmed = self::independentlyTerminal($state);
            $producer = in_array($state['producer_terminal_status'] ?? null, ['stopped', 'succeeded', 'failed'], true)
                ? $state['producer_terminal_status'] : '';
            $persisted = in_array($state['persisted_run_status'] ?? null,
                ['scheduled', 'running', 'succeeded', 'failed', 'stopped', 'partial-succeeded', 'paused'], true) ? $state['persisted_run_status'] : '';
            $conflict = !empty($state['lifecycle_conflict']);
            foreach ($state['completed_generations'] ?? [] as $completed) $conflict = $conflict || !empty($completed['lifecycle_conflict']);
            foreach ($state['handoff_cleanup'] ?? [] as $cleanup) {
                $conflict = $conflict || !empty($cleanup['lifecycle_conflict']);
                foreach ($cleanup['completed_generations'] ?? [] as $completed) $conflict = $conflict || !empty($completed['lifecycle_conflict']);
                if (is_array($cleanup['upstream'] ?? null) && !self::independentlyTerminal($cleanup)) {
                    ++$pending;
                    $stopConfirmed = $stopConfirmed && !empty($cleanup['upstream_stop_confirmed']);
                    $terminalConfirmed = false;
                }
            }
            $submissions = self::submissionCounts($state);
            $pending += $submissions['additional_pending'];
            if ($submissions['pending'] > 0) $terminalConfirmed = false;
            return [
            'status' => $state['status'] ?? 'pending',
            'cancel_requested' => !empty($state['cancel_requested']),
            'timeout_reached' => !empty($state['timeout_reached']),
            'transport_closed' => !empty($state['transport_closed']),
            'upstream_stop_confirmed' => $stopConfirmed,
            'upstream_terminal_confirmed' => $terminalConfirmed,
            'upstream_terminal_status' => $terminalConfirmed ? (string) ($state['upstream_terminal_status'] ?? '') : '',
            'upstream_pending_task_count' => $pending,
            'producer_terminal_received' => $producer !== '',
            'producer_terminal_status' => $producer,
            'producer_terminal_observed_at' => self::observationTime($state['producer_terminal_observed_at'] ?? null),
            'persisted_run_status' => $persisted,
            'persisted_run_observed_at' => self::observationTime($state['persisted_run_observed_at'] ?? null),
            'stop_accepted_at' => self::observationTime($state['stop_accepted_at'] ?? null),
            'generation_transfer_closed' => !empty($state['transfer_closed']),
            'generation_transfer_closed_at' => self::observationTime($state['transfer_closed_at'] ?? null),
            'generation_verification_elapsed_ms' => max(0, (int) ($state['generation_verification_elapsed_ms'] ?? 0)),
            'lifecycle_conflict' => $conflict,
            'submitted_attempt_count' => $submissions['total'],
            'submitted_transfer_closed_count' => $submissions['closed'],
            'submitted_pending_attempt_count' => $submissions['pending'],
            'submitted_without_identity_count' => $submissions['unidentified'],
            'pre_execution_closed_attempt_count' => $submissions['pre_execution'],
            'not_submitted_attempt_count' => $submissions['not_submitted'],
            'rejected_before_execution_attempt_count' => $submissions['rejected_before_execution'],
            ];
        }, false);
    }

    private function update(callable $apply, bool $persist = true)
    {
        $guard = @fopen($this->directory . DIRECTORY_SEPARATOR . '.gc.lock', 'c+');
        if ($guard === false || !flock($guard, LOCK_SH)) {
            if (is_resource($guard)) fclose($guard);
            throw new \RuntimeException('Assistant request control unavailable');
        }
        $handle = @fopen($this->path, 'c+');
        if ($handle === false || !flock($handle, $persist ? LOCK_EX : LOCK_SH)) {
            if (is_resource($handle)) fclose($handle);
            flock($guard, LOCK_UN);
            fclose($guard);
            throw new \RuntimeException('Assistant request control unavailable');
        }
        try {
            @chmod($this->path, 0600);
            $raw = stream_get_contents($handle);
            $state = is_string($raw) ? json_decode($raw, true) : null;
            if ($raw !== '' && !is_array($state)) throw new \RuntimeException('Assistant request control unavailable');
            if (!is_array($state) || ((int) ($state['expires_at'] ?? 0) <= time() && !self::hasUnconfirmedIdentity($state))) {
                $state = ['expires_at' => time() + $this->ttl];
            }
            $result = $apply($state);
            if (!$persist) return $result;
            $state['expires_at'] = time() + $this->ttl;
            $encoded = json_encode($state, JSON_THROW_ON_ERROR);
            rewind($handle);
            if (!ftruncate($handle, 0) || fwrite($handle, $encoded) !== strlen($encoded) || !fflush($handle)) {
                throw new \RuntimeException('Assistant request control unavailable');
            }
            return $result;
        } finally {
            flock($handle, LOCK_UN);
            fclose($handle);
            flock($guard, LOCK_UN);
            fclose($guard);
        }
    }

    private static function hasUnconfirmedIdentity(array $state): bool
    {
        if (self::submissionCounts($state)['pending'] > 0) return true;
        if (is_array($state['upstream'] ?? null) && !self::independentlyTerminal($state)) return true;
        foreach ($state['handoff_cleanup'] ?? [] as $cleanup) {
            if (is_array($cleanup['upstream'] ?? null) && !self::independentlyTerminal($cleanup)) return true;
        }
        return false;
    }

    /** The global guard prevents unlink/reopen races with any other worker. */
    private function collectExpired(): void
    {
        $guard = @fopen($this->directory . DIRECTORY_SEPARATOR . '.gc.lock', 'c+');
        if ($guard === false) return;
        try {
            if (!flock($guard, LOCK_EX | LOCK_NB)) return;
            $paths = glob($this->directory . DIRECTORY_SEPARATOR . '*.json') ?: [];
            // Bound each sweep; rotating the scan also reclaims older files in large directories.
            if (count($paths) > 256) shuffle($paths);
            foreach (array_slice($paths, 0, 256) as $path) {
                if ((int) @filemtime($path) > time() - 600) continue;
                $state = json_decode((string) @file_get_contents($path), true);
                // Corrupt controls fail closed; their identity cannot be inferred.
                if (is_array($state) && (int) ($state['expires_at'] ?? 0) <= time()
                    && !self::hasUnconfirmedIdentity($state)) @unlink($path);
            }
        } finally {
            flock($guard, LOCK_UN);
            fclose($guard);
        }
    }
}