<?php
// SPDX-License-Identifier: GPL-3.0-or-later
declare(strict_types=1);

namespace Aiwend\Chat;

use Aiwend\Audit\AuditLog;
use Aiwend\Db\Connection;
use Aiwend\Knowledge\Grounding;
use Aiwend\Knowledge\RetrievedChunk;
use Aiwend\Policy\Capabilities;
use Aiwend\Policy\Policy;
use Aiwend\Policy\Principal;
use Aiwend\Provider\CancellationToken;
use Aiwend\Provider\ModelAdapter;
use Aiwend\Provider\ModelRequest;
use Aiwend\Provider\ProviderException;
use Aiwend\Support\Clock;
use Aiwend\Support\RequestId;
use Aiwend\Usage\BudgetExceededException;
use Aiwend\Usage\ClientAllowance;
use Aiwend\Usage\ExhaustionAlerts;
use Aiwend\Usage\Ledger;
use Aiwend\Usage\Reservation;
use Aiwend\Usage\SpendingHaltedException;

/**
 * Chat orchestration (C01). Order of operations guarantees that nothing paid
 * is dispatched unless policy, validation and budget reservation all pass:
 *
 *   policy → validate → replay check → reserve → store user turn → dispatch → settle/release/uncertain
 *
 * The orchestrator depends only on the ModelAdapter contract; adapters are
 * resolved per bot by the injected resolver.
 */
final class ChatService
{
    public const MAX_MESSAGE_CHARS = 4000;
    /**
     * Dormant: genuine embedding-budget exhaustion during retrieval would be the public 429 "usage_limit_reached" instead of the
     * generic 503. Remote embeddings are not wired, so lexical retrieval can never reach it; flipping this needs owner approval.
     */
    private const EMBEDDING_BUDGET_AS_429 = false;
    /** How an assistant response ended (messages.outcome; NULL on pre-v7 rows means completed). */
    public const OUTCOME_COMPLETED = 'completed';
    public const OUTCOME_STOPPED = 'stopped';
    /** Stage D: a grounded assistant stored its fixed no-source answer because retrieval found nothing qualifying, or because the evidence was withdrawn before dispatch. */
    public const OUTCOME_NO_MATCH = 'no_match';
    public const OUTCOME_WITHDRAWN = 'withdrawn';
    public const HISTORY_TURNS = 12;
    /**
     * Phase 96: stands in for a stored answer that cited a source which is no longer public and ready (made private, deleted). Used in the
     * history sent to the provider, in replays and in a visitor's own transcript. The stored row is never changed or deleted.
     */
    public const WITHHELD = 'This earlier answer is no longer available because its source was withdrawn.';

    /** @var callable(array<string, mixed>): ModelAdapter */
    private $adapterResolver;

    /** @param callable(array<string, mixed>): ModelAdapter $adapterResolver */
    public function __construct(
        private Connection $db,
        private Policy $policy,
        private SessionService $sessions,
        private BotRepository $bots,
        private Ledger $ledger,
        private AuditLog $audit,
        private Clock $clock,
        callable $adapterResolver,
        private int $sessionBudgetUnits = 20000,
        private ?Grounding $grounding = null,
        ?OutputGuard $guard = null,
        private ?\Closure $knowledgeOff = null,
        // Effective provider model (e.g. the site default for assistants that inherit it).
        private ?\Closure $modelResolver = null,
        private ?ExhaustionAlerts $alerts = null,
        private ?\Aiwend\Usage\ProviderBreaker $breaker = null,
        private ?\Aiwend\Usage\ConnectionHealth $health = null,
    ) {
        $this->adapterResolver = $adapterResolver;
        $this->guard = $guard ?? new OutputGuard([]);
    }

    private OutputGuard $guard;

    /** @var \Closure(array<string, mixed>): ?string|null */
    private ?\Closure $providerGate = null;

    /**
     * Release 0.9.0: a server-side check run for every NON-preview request, after the assistant is loaded and before anything is retrieved,
     * reserved or dispatched. It receives the assistant, resolves the CURRENT effective provider itself and returns an error code to refuse
     * with, or null to continue. A refusal creates no reservation and no outbound request.
     *
     * @param \Closure(array<string, mixed>): ?string $gate
     */
    public function withProviderGate(\Closure $gate): self
    {
        $this->providerGate = $gate;
        return $this;
    }

    /**
     * @param callable(string): void|null $onDelta
     * @throws ChatException
     */
    /**
     * $preview: a private preview by an operator (the caller has already checked the preview
     * permission). It may use an assistant that is not Active; every other rule still applies.
     *
     * $client: the visitor's daily budget bucket (Phase 89). It is charged together with the conversation and the site bucket in
     * one atomic reservation, so opening more conversations never creates more allowance. Null for operator previews.
     */
    public function send(Principal $principal, int $sessionId, string $message, string $requestId, ?CancellationToken $cancel = null, ?callable $onDelta = null, bool $preview = false, ?ClientAllowance $client = null): ChatReply
    {
        $cancel ??= CancellationToken::none();
        $siteId = $principal->siteId;

        $session = $this->sessions->resource($siteId, $sessionId);
        $decision = $this->policy->check($principal, 'chat.message.send', $session);
        if (!$decision->allowed) {
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ChatException($decision->code, $decision->httpStatus());
        }
        if ($session === null || $session->attr('active') !== true) {
            throw new ChatException('not_found', 404);
        }
        if (!RequestId::isValid($requestId)) {
            throw new ChatException('invalid_request_id', 400);
        }
        $message = $this->validateMessage($message);

        // Replay: the same request ID returns the stored reply without dispatching again.
        $replay = $this->storedReply($siteId, $sessionId, $requestId);
        if ($replay !== null) {
            return $replay;
        }

        $bot = $this->bots->find($siteId, (int) $session->attr('bot_id'));
        if ($bot === null || ($bot['status'] !== 'active' && !$preview)) {
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ChatException('bot_unavailable', 409, $requestId);
        }
        // Provider gate (0.9.0): refused before any retrieval, reservation or provider request. Operator previews are exempt: the caller has
        // already checked the preview permission, and an operator is not an anonymous visitor.
        if (!$preview && $this->providerGate !== null) {
            $refusal = ($this->providerGate)($bot);
            if ($refusal !== null) {
                // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
                throw new ChatException($refusal, 403, $requestId);
            }
        }
        // An assistant that uses knowledge never falls back to answering without it while the
        // Knowledge module is off; it refuses until the module is back.
        if ($this->grounding === null && $this->knowledgeOff !== null && ($this->knowledgeOff)($siteId, (int) $bot['id'])) {
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ChatException('knowledge_unavailable', 503, $requestId);
        }

        $botId = (int) $bot['id'];
        // Grounded bots answer only from authorized sources. With no qualifying evidence the
        // answer is a fixed "not found" message and nothing is dispatched or charged.
        $grounded = $this->grounding !== null && $this->grounding->enabledFor($siteId, $botId);
        // A metered (remote) query embedding is reserved against the site AND the session bucket, so that bucket must exist
        // before retrieval. It is created exactly as before, under the session row lock, and only for bots whose retrieval is
        // metered; lexical retrieval keeps creating it at the chat reservation, as it always did.
        if ($grounded && $this->grounding->usesMeteredEmbeddings() && !$this->ensureSessionBucket($siteId, $sessionId)) {
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ChatException('not_found', 404, $requestId);
        }
        try {
            $chunks = $grounded ? $this->grounding->retrieve($principal, $botId, $message, $requestId) : [];
        } catch (\Throwable $e) {
            $this->knowledgeFailed($e, $principal, $sessionId, $requestId); // nothing was reserved or dispatched yet
        }
        if ($grounded && $chunks === []) {
            return $this->abstain($principal, $sessionId, $botId, $message, $requestId, 'no_evidence');
        }

        $adapter = ($this->adapterResolver)($bot);
        $history = array_merge($this->history($siteId, $sessionId, $message), [['role' => 'user', 'content' => $message]]);
        $request = $this->buildRequest($bot, $adapter, $history, $requestId, $grounded, $chunks);

        $units = ($adapter instanceof \Aiwend\Provider\RequestBudget ? $adapter->inputReservation($request) : $request->estimatedInputTokens()) + $request->maxOutputTokens;
        $buckets = ['site', 'session:' . $sessionId];
        if (!$this->ensureSessionBucket($siteId, $sessionId)) {
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ChatException('not_found', 404, $requestId);
        }
        if ($client !== null) {
            // Created only if absent (one statement, deadlock-free) and never changed afterwards: new settings apply from the
            // next day's bucket. The Ledger locks client, conversation and site rows in one fixed order. A new bucket always holds
            // room for two of this assistant's answers, so a small limit next to a long maximum answer never refuses a visitor outright.
            $this->ledger->ensureBucket($siteId, $client->bucket, static fn(): int => max($client->limitUnits(), 2 * $units));
            $buckets[] = $client->bucket;
        }
        try {
            $reservation = $this->ledger->reserve($siteId, $requestId, $buckets, $units);
        } catch (SpendingHaltedException) {
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ChatException('spending_paused', 503, $requestId);
        } catch (BudgetExceededException $e) {
            if ($this->sessions->resource($siteId, $sessionId) === null) {
                // Erased between creating the bucket and reserving: the erase removed the bucket.
                // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
                throw new ChatException('not_found', 404, $requestId);
            }
            $this->audit->record($siteId, $principal->actorRef(), 'chat.send', "session:$sessionId", 'budget_denied', $requestId, $e->bucket);
            $this->alerts?->note($e->bucket);
            $code = $this->ledger->balance($siteId, $e->bucket) === null
                ? 'budget_not_configured'
                : (str_starts_with($e->bucket, 'client:') ? 'client_limit_reached' : 'usage_limit_reached');
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ChatException($code, 429, $requestId);
        }
        if ($reservation->duplicate) {
            // Same request ID is in flight or ended without a stored reply; never dispatch twice.
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ChatException($reservation->state === Reservation::UNCERTAIN ? 'outcome_uncertain' : 'duplicate_request', 409, $requestId);
        }
        // Stored evidence of where this usage comes from (Usage page: by connection and assistant).
        $this->ledger->attribute(
            $siteId,
            $requestId,
            ($adapter->capabilities()['simulated'] ?? false) ? Reservation::ORIGIN_SIMULATED : Reservation::ORIGIN_PROVIDER,
            $botId,
            $adapter instanceof \Aiwend\Provider\ConnectionBound ? $adapter->connectionId() : null,
            $request->model
        );
        // Circuit breaker for repeated AMBIGUOUS provider failures (outage protection). It runs after our own reservation row exists, so
        // concurrent requests see each other. A refused request is released at once: it costs the visitor and the site nothing, and
        // nothing already held is touched.
        $breakerConnection = $adapter instanceof \Aiwend\Provider\ConnectionBound ? $adapter->connectionId() : null;
        if ($this->breaker !== null && $breakerConnection !== null && !($adapter->capabilities()['simulated'] ?? false)) {
            // The source is who answers for this request: the visitor's client bucket, else the conversation (Phase 93: one visitor
            // can pause only themselves, never the shared connection for everyone).
            $block = $this->breaker->check($siteId, $breakerConnection, $client?->bucket ?? 'session:' . $sessionId, 'session:' . $sessionId);
            if ($block !== null) {
                $this->ledger->release($siteId, $requestId, 'provider-breaker-' . $block['reason']);
                $this->audit->record($siteId, $principal->actorRef(), 'chat.send', "session:$sessionId", 'breaker_' . $block['reason'], $requestId, 'connection ' . $breakerConnection);
                // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
                throw new ChatException('provider_cooling_down', 503, $requestId);
            }
        }

        try {
            $stored = $this->appendMessage($siteId, $sessionId, (int) $bot['id'], 'user', $message, $requestId);
        } catch (\Throwable $e) {
            $this->ledger->release($siteId, $requestId, 'store-failed');
            throw $e;
        }
        if ($stored === 0) {
            // Session erased between authorization and storing the question: nothing dispatched.
            $this->ledger->release($siteId, $requestId, 'session-erased');
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ChatException('not_found', 404, $requestId);
        }

        if ($cancel->isCancelled()) {
            $this->ledger->release($siteId, $requestId, 'cancelled-before-dispatch');
            $this->removeUnansweredTurn($siteId, $sessionId, $stored);
            return new ChatReply($requestId, '', true, false);
        }

        if ($grounded) {
            // Re-validate source permissions immediately before content leaves for the model.
            try {
                $checked = $this->grounding->recheck($siteId, $botId, $chunks);
            } catch (\Throwable $e) {
                $this->ledger->release($siteId, $requestId, 'knowledge-check-failed'); // never dispatched: the hold is returned
                $this->knowledgeFailed($e, $principal, $sessionId, $requestId);
            }
            if ($checked === []) {
                $this->ledger->release($siteId, $requestId, 'evidence-revoked');
                return $this->storeAbstention($principal, $sessionId, $botId, $requestId, 'evidence_revoked');
            }
            if (count($checked) !== count($chunks)) {
                $chunks = $checked;
                $request = $this->buildRequest($bot, $adapter, $history, $requestId, true, $chunks);
            }
        }

        $this->ledger->markDispatched($siteId, $requestId);
        try {
            $streamGuard = $onDelta === null ? null : new StreamGuard($this->guard, $onDelta);
            $result = $adapter->complete($request, $cancel, $streamGuard === null ? null : [$streamGuard, 'push']);
            $streamGuard?->flush();
        } catch (ProviderException $e) {
            $this->handleProviderFailure($principal, $sessionId, $requestId, $e, self::usageBasis($adapter), ($adapter->capabilities()['simulated'] ?? false) ? null : $breakerConnection);
            if (!$e->dispatched || $e->rejected) {
                // Nothing reached the provider (or it refused before processing): the question was never answered, so it is
                // not kept in the transcript (a retry would otherwise store it twice).
                $this->removeUnansweredTurn($siteId, $sessionId, $stored);
            }
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ChatException('provider_' . $e->kind, $e->kind === ProviderException::RATE_LIMITED ? 429 : 502, $requestId);
        } catch (\Throwable $e) {
            $this->ledger->markUncertain($siteId, $requestId, 'unexpected-after-dispatch');
            $this->audit->record($siteId, $principal->actorRef(), 'chat.send', "session:$sessionId", 'uncertain', $requestId, $e::class);
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ChatException('internal_error', 500, $requestId);
        }

        if ($result->usageUnknown) {
            // Stopped after dispatch (F3): the visitor keeps the text streamed so far, but the provider's usage is unknown and billing may continue there, so the
            // hold stays UNCERTAIN (never shown as settled) and nothing is retried. A stop is not proof that the connection answers, so the breaker is not told.
            $this->ledger->markUncertain($siteId, $requestId, 'user-stopped');
        } else {
            $this->ledger->settle($siteId, $requestId, $result->totalTokens(), $result->providerRef, self::usageBasis($adapter), $result->inputTokens, $result->outputTokens, $result->detail);
        }
        if (!$result->usageUnknown && $this->breaker !== null && $breakerConnection !== null && !($adapter->capabilities()['simulated'] ?? false)) {
            $this->breaker->recordSuccess($siteId, $breakerConnection); // proof that the connection is answering (see ProviderBreaker)
        }
        $text = $result->text;
        // Model output is untrusted. Tool calls it proposes are never executed from chat;
        // they are only recorded (no tool is enabled for chat in the pilot).
        if (preg_match('/<\/?tool_call|"(tool|function_call|tool_calls)"\s*:|\[\[tool:/i', $text)) {
            $this->audit->record($siteId, $principal->actorRef(), 'chat.tool_call_ignored', "session:$sessionId", 'ignored', $requestId);
        }
        $guarded = $this->guard->sanitize($text);
        $text = $guarded['text'];
        if ($guarded['removed'] > 0) {
            $this->audit->record($siteId, $principal->actorRef(), 'chat.output_guarded', "session:$sessionId", 'removed', $requestId, (string) $guarded['removed']);
        }
        $citations = [];
        if ($grounded) {
            $validated = Grounding::validate($this->db, $siteId, $botId, Grounding::resolve($text, $chunks));
            $text = $validated['text'];
            $citations = $validated['citations'];
            if ($validated['dropped'] > 0) {
                $this->audit->record($siteId, $principal->actorRef(), 'chat.citation_dropped', "session:$sessionId", 'invalid_provenance', $requestId, (string) $validated['dropped']);
            }
        }
        // Erasure safety: the answer is stored only if the session still exists at the moment of
        // writing (checked under the session row lock). Usage was already settled above, so the
        // provider work is accounted for even when the answer is discarded.
        $erased = $text !== ''
            ? $this->appendMessage($siteId, $sessionId, $botId, 'assistant', $text, $requestId, $citations, $result->cancelled ? self::OUTCOME_STOPPED : self::OUTCOME_COMPLETED, $grounded ? array_values(array_unique(array_map(static fn(RetrievedChunk $c): int => $c->sourceId, $chunks))) : []) === 0
            : $this->sessions->resource($siteId, $sessionId) === null;
        if ($erased) {
            $this->audit->record($siteId, $principal->actorRef(), 'chat.send', "session:$sessionId", 'discarded_session_erased', $requestId);
            return new ChatReply($requestId, '', true, false);
        }
        $this->audit->record($siteId, $principal->actorRef(), 'chat.send', "session:$sessionId", $result->cancelled ? 'cancelled' : 'ok', $requestId);
        // Internal ids (source/version/chunk) are stored, not returned to the client.
        $public = array_map(static fn(array $c): array => ['n' => $c['n'], 'title' => $c['title'], 'url' => $c['url'], 'anchor' => $c['anchor'], 'provenance' => 'verified', 'support' => 'not_evaluated', 'stale' => false], $citations);
        return new ChatReply($requestId, $text, $result->cancelled, false, $public);
    }

    /**
     * Transcript for the session owner (or an authorized exporter).
     *
     * @return list<array{role:string, content:string, request_id:string, created_at:int, citations:list<array<string,mixed>>}>
     */
    public function transcript(Principal $principal, int $sessionId): array
    {
        $decision = $this->policy->check($principal, 'chat.history.read', $this->sessions->resource($principal->siteId, $sessionId));
        if (!$decision->allowed) {
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new ChatException($decision->code, $decision->httpStatus());
        }
        $rows = $this->db->select(
            'SELECT id, role, content, request_id, created_at, outcome FROM {messages} WHERE site_id = ? AND session_id = ? ORDER BY seq',
            [$principal->siteId, $sessionId]
        );
        // A visitor reading a conversation back does not get an answer whose source was withdrawn since (fail closed for that message only);
        // an authorized exporter reads the stored record unchanged, for audit.
        $withdrawn = $principal->can(Capabilities::EXPORT_CONVERSATIONS) ? [] : $this->withdrawnAnswers($principal->siteId, array_map(static fn(array $r): int => (int) $r['id'], array_filter($rows, static fn(array $r): bool => $r['role'] === 'assistant')));
        $out = [];
        foreach ($rows as $i => $r) {
            $assistant = $r['role'] === 'assistant';
            $hidden = $assistant && isset($withdrawn[(int) $r['id']]);
            $out[] = [
                'role' => (string) $r['role'],
                'content' => $hidden ? self::WITHHELD : (string) $r['content'],
                'request_id' => (string) $r['request_id'],
                'created_at' => (int) $r['created_at'],
                // assistant: completed | stopped. user: answered | unanswered (no answer was stored after it,
                // e.g. the request failed or the conversation was interrupted).
                'outcome' => $assistant
                    ? ($r['outcome'] === self::OUTCOME_STOPPED ? self::OUTCOME_STOPPED : self::OUTCOME_COMPLETED)
                    : (($rows[$i + 1]['role'] ?? null) === 'assistant' ? 'answered' : 'unanswered'),
                'rating' => (int) ($this->db->selectOne('SELECT rating FROM {message_feedback} WHERE site_id = ? AND message_id = ?', [$principal->siteId, (int) $r['id']])['rating'] ?? 0),
                'citations' => $assistant && !$hidden ? $this->citationsFor($principal->siteId, (int) $r['id']) : [],
            ];
        }
        return $out;
    }

    /**
     * Assistant messages that cite at least one source which is no longer public and ready (or no longer exists).
     * Only citations identify the source of an answer: text a model produced WITHOUT citing it cannot be traced back (documented limit).
     *
     * @param list<int> $messageIds
     * @return array<int, true> message id => true
     */
    private function withdrawnAnswers(int $siteId, array $messageIds): array
    {
        if ($messageIds === []) {
            return [];
        }
        $in = implode(', ', array_fill(0, count($messageIds), '?'));
        $out = [];
        foreach ($this->db->select(
            "SELECT DISTINCT mc.message_id FROM {message_citations} mc LEFT JOIN {knowledge_sources} s ON s.id = mc.source_id AND s.site_id = mc.site_id
             WHERE mc.site_id = ? AND mc.message_id IN ($in) AND (s.id IS NULL OR s.visibility <> 'public' OR s.status <> 'ready')",
            array_merge([$siteId], $messageIds)
        ) as $row) {
            $out[(int) $row['message_id']] = true;
        }
        // Provenance (messages.grounding): every source whose text was sent for that answer, cited or not. This is what catches an answer
        // that quotes a source without a [n] marker, or whose citation was dropped as invalid. NULL = written before provenance existed.
        $legacy = [];
        $sourceIds = [];
        $grounding = [];
        foreach ($this->db->select("SELECT id, bot_id, content, grounding FROM {messages} WHERE site_id = ? AND id IN ($in) AND role = 'assistant'", array_merge([$siteId], $messageIds)) as $m) {
            if ($m['grounding'] === null) {
                if ((string) $m['content'] !== Grounding::ABSTAIN_TEXT) {
                    $legacy[(int) $m['id']] = (int) $m['bot_id'];
                }
                continue;
            }
            $ids = array_values(array_filter(array_map('intval', explode(',', (string) $m['grounding']))));
            $grounding[(int) $m['id']] = $ids;
            $sourceIds = array_merge($sourceIds, $ids);
        }
        $closed = [];
        if ($sourceIds !== []) {
            $sourceIds = array_values(array_unique($sourceIds));
            $sin = implode(', ', array_fill(0, count($sourceIds), '?'));
            $open = [];
            foreach ($this->db->select("SELECT id FROM {knowledge_sources} WHERE site_id = ? AND id IN ($sin) AND visibility = 'public' AND status = 'ready'", array_merge([$siteId], $sourceIds)) as $s) {
                $open[(int) $s['id']] = true;
            }
            foreach ($grounding as $mid => $ids) {
                foreach ($ids as $sid) {
                    if (!isset($open[$sid])) {
                        $out[$mid] = true;
                    }
                }
            }
        }
        // Answers stored before provenance existed: fail closed when the assistant now has ANY attached source that is not public and ready
        // (the answer may have quoted it without citing it). Only what is sent onward is replaced; nothing is deleted.
        foreach (array_unique($legacy) as $botId) {
            $closed[$botId] = $this->db->selectOne(
                "SELECT s.id FROM {bot_sources} b JOIN {knowledge_sources} s ON s.id = b.source_id AND s.site_id = ? WHERE b.bot_id = ? AND (s.visibility <> 'public' OR s.status <> 'ready') LIMIT 1",
                [$siteId, $botId]
            ) !== null;
        }
        foreach ($legacy as $mid => $botId) {
            if ($closed[$botId]) {
                $out[$mid] = true;
            }
        }
        return $out;
    }

    /** Recent turns sent to the model as context. @return list<array{role:string, content:string}> */
    private function history(int $siteId, int $sessionId, ?string $retryOf = null): array
    {
        // Stopped (incomplete) answers are never given to the model as if they were complete answers.
        $rows = $this->db->select(
            "SELECT id, role, content FROM {messages} WHERE site_id = ? AND session_id = ? AND (outcome IS NULL OR outcome <> 'stopped') ORDER BY seq DESC LIMIT ?",
            [$siteId, $sessionId, self::HISTORY_TURNS]
        );
        // Phase 96: an earlier answer that cited a source which is no longer public is NOT sent to the provider again; a neutral line takes its
        // place so the turns keep alternating. What was already sent in earlier requests cannot be recalled; this controls every future one.
        $withdrawn = $this->withdrawnAnswers($siteId, array_map(static fn(array $r): int => (int) $r['id'], array_filter($rows, static fn(array $r): bool => $r['role'] === 'assistant')));
        $turns = array_map(static fn(array $r): array => ['role' => (string) $r['role'], 'content' => isset($withdrawn[(int) $r['id']]) ? self::WITHHELD : (string) $r['content']], array_reverse($rows));
        if ($retryOf !== null && $turns !== [] && end($turns) === ['role' => 'user', 'content' => $retryOf]) {
            array_pop($turns); // A retried question (unanswered or stopped) is sent once, not twice.
        }
        return $turns;
    }

    /**
     * Basis of the usage an adapter returns: provider-reported only if it declares so; the mock is
     * simulated; anything else (an adapter that counts locally) is an unconfirmed estimate.
     */
    private function removeUnansweredTurn(int $siteId, int $sessionId, int $messageId): void
    {
        $this->db->execute("DELETE FROM {messages} WHERE id = ? AND site_id = ? AND session_id = ? AND role = 'user'", [$messageId, $siteId, $sessionId]);
    }

    private static function usageBasis(ModelAdapter $adapter): string
    {
        $caps = $adapter->capabilities();
        if (($caps['simulated'] ?? false) === true) {
            return Reservation::BASIS_SIMULATED;
        }
        return ($caps['usage'] ?? '') === 'reported' ? Reservation::BASIS_REPORTED : Reservation::BASIS_ESTIMATED;
    }

    private function handleProviderFailure(Principal $principal, int $sessionId, string $requestId, ProviderException $e, string $basis, ?int $connectionId = null): void
    {
        $siteId = $principal->siteId;
        if (!$e->dispatched) {
            // markDispatched already ran; move back is not allowed, so settle at zero instead.
            $this->ledger->settle($siteId, $requestId, 0, null, Reservation::BASIS_NOT_DISPATCHED);
            $outcome = 'failed_not_dispatched';
        } elseif ($e->rejected) {
            // A definitive pre-processing refusal (401, 403, 404, 429 ...): nothing was generated, so the hold is returned at once
            // instead of staying uncertain. Settling is exactly-once (state guard), so a later reconcile cannot release it again.
            $this->ledger->settle($siteId, $requestId, 0, null, Reservation::BASIS_REJECTED);
            $outcome = 'failed_rejected';
        } elseif ($e->usageTokens !== null) {
            $this->ledger->settle($siteId, $requestId, $e->usageTokens, null, $basis);
            $outcome = 'failed_usage_known';
        } else {
            $this->ledger->markUncertain($siteId, $requestId, 'provider-' . $e->kind);
            $outcome = 'uncertain';
        }
        $this->audit->record($siteId, $principal->actorRef(), 'chat.send', "session:$sessionId", $outcome, $requestId, 'provider ' . $e->kind . ($e->httpStatus !== null ? ' http ' . $e->httpStatus : ''));
        $this->observeFailure($connectionId, $e);
    }

    /**
     * The administrator's view of what live traffic is meeting (Diagnostics, Overview, Connections): one bounded fact per connection and
     * category. Best effort by design: this runs after the ledger and the log are settled, and nothing it does or throws can change the
     * outcome of the request. It only reads what is already known (the exception's kind, status and dispatch flags).
     */
    private function observeFailure(?int $connectionId, ProviderException $e): void
    {
        if ($this->health === null || $connectionId === null) {
            return;
        }
        try {
            $this->health->record(
                $connectionId,
                \Aiwend\Support\FailureCatalog::categoryForProvider($e->kind, $e->httpStatus),
                $e->httpStatus,
                \Aiwend\Support\FailureCatalog::billingFor($e->dispatched, $e->rejected, $e->usageTokens !== null)
            );
        } catch (\Throwable) {
            // a fact that could not be stored is a fact the administrator will see on the next failure
        }
    }

    private function storedReply(int $siteId, int $sessionId, string $requestId): ?ChatReply
    {
        $row = $this->db->selectOne(
            'SELECT id, content, outcome FROM {messages} WHERE site_id = ? AND session_id = ? AND request_id = ? AND role = ?',
            [$siteId, $sessionId, $requestId, 'assistant']
        );
        if ($row === null) {
            return null;
        }
        if ($this->withdrawnAnswers($siteId, [(int) $row['id']]) !== []) {
            // The source was withdrawn after the answer was stored: a replay must not serve it (or its citations) again.
            return new ChatReply($requestId, self::WITHHELD, false, true, [], true);
        }
        return new ChatReply($requestId, (string) $row['content'], $row['outcome'] === self::OUTCOME_STOPPED, true, $this->citationsFor($siteId, (int) $row['id']));
    }

    private function validateMessage(string $message): string
    {
        if (!mb_check_encoding($message, 'UTF-8')) {
            throw new ChatException('invalid_message', 400);
        }
        // Strip control characters except newline and tab.
        $message = trim((string) preg_replace('/[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]/u', '', $message));
        if ($message === '') {
            throw new ChatException('empty_message', 400);
        }
        if (mb_strlen($message) > self::MAX_MESSAGE_CHARS) {
            throw new ChatException('message_too_long', 413);
        }
        return $message;
    }

    /**
     * @param array<string, mixed> $bot
     * @param list<array{role:string, content:string}> $history
     * @param list<RetrievedChunk> $chunks
     */
    private function buildRequest(array $bot, ModelAdapter $adapter, array $history, string $requestId, bool $grounded, array $chunks): ModelRequest
    {
        return new ModelRequest(
            $this->modelResolver !== null ? (string) ($this->modelResolver)($bot) : (string) $bot['model'],
            // Server-owned: bot instruction plus, for grounded bots, the fixed grounding rules.
            (string) $bot['instruction'] . ($grounded ? "\n\n" . Grounding::RULES : ''),
            $history,
            min((int) $bot['max_output_tokens'], $adapter->capabilities()['max_output_tokens']),
            $requestId,
            60,
            $grounded ? Grounding::context($chunks) : [],
        );
    }

    /** No qualifying evidence before any reservation: store the turn and a fixed answer; nothing is dispatched. */
    /**
     * One bounded outcome for every failure of knowledge retrieval (embedding provider, budget, database, anything):
     * the audit log gets a stable reason code only (never the exception text, which could carry provider detail), the
     * visitor gets a generic 503 "knowledge_failed" with the request ID ("knowledge_unavailable" stays the Knowledge-module-off case). There is no fallback to another
     * retrieval method, no stored "not found" answer (nothing was searched), no model call and no retry.
     */
    private function knowledgeFailed(\Throwable $e, Principal $principal, int $sessionId, string $requestId): never
    {
        $reason = $e instanceof \Aiwend\Knowledge\ExtractionException && preg_match('/^[a-z_]{1,40}$/', $e->reason) === 1 ? $e->reason : 'retrieval_failed';
        $failure = \Aiwend\Knowledge\RetrievalFailure::classify($e);
        // The audit row keeps its reason code and now carries the typed class (a fixed word, never exception text).
        $this->audit->record($principal->siteId, $principal->actorRef(), 'chat.knowledge_failed', "session:$sessionId", $reason, $requestId, $failure->value);
        $public = $failure->publicError(self::EMBEDDING_BUDGET_AS_429);
        // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
        throw new ChatException($public['code'], $public['status'], $requestId, $failure->value);
    }

    private function abstain(Principal $principal, int $sessionId, int $botId, string $message, string $requestId, string $reason): ChatReply
    {
        $this->appendMessage($principal->siteId, $sessionId, $botId, 'user', $message, $requestId);
        return $this->storeAbstention($principal, $sessionId, $botId, $requestId, $reason);
    }

    private function storeAbstention(Principal $principal, int $sessionId, int $botId, string $requestId, string $reason): ChatReply
    {
        $this->appendMessage($principal->siteId, $sessionId, $botId, 'assistant', Grounding::ABSTAIN_TEXT, $requestId, [], $reason === 'evidence_revoked' ? self::OUTCOME_WITHDRAWN : self::OUTCOME_NO_MATCH, []);
        $this->audit->record($principal->siteId, $principal->actorRef(), 'chat.send', "session:$sessionId", 'abstained', $requestId, $reason);
        return new ChatReply($requestId, Grounding::ABSTAIN_TEXT, false, false, [], true);
    }

    /** @param list<array{n:int, source_id:int, version_id:int, chunk_id:int, title:string, url:?string, anchor:string}> $citations */
    private function storeCitations(int $siteId, int $messageId, array $citations): void
    {
        foreach ($citations as $c) {
            $this->db->insert('message_citations', [
                'site_id' => $siteId, 'message_id' => $messageId, 'n' => $c['n'], 'source_id' => $c['source_id'],
                'version_id' => $c['version_id'], 'chunk_id' => $c['chunk_id'], 'anchor' => mb_substr($c['anchor'], 0, 191),
                'title' => mb_substr($c['title'], 0, 191), 'url' => $c['url'],
            ]);
        }
    }

    /**
     * Stored citations. provenance "verified" = validated against source/version/anchor when the
     * answer was produced; support is never evaluated by this system; stale = the source has a
     * newer version, or was unpublished or deleted, since then.
     *
     * @return list<array{n:int, title:string, url:?string, anchor:string, provenance:string, support:string, stale:bool}>
     */
    private function citationsFor(int $siteId, int $messageId): array
    {
        return array_map(static fn(array $r): array => [
            'n' => (int) $r['n'], 'title' => (string) $r['title'], 'url' => $r['url'] === null ? null : (string) $r['url'], 'anchor' => (string) $r['anchor'],
            'provenance' => 'verified', 'support' => 'not_evaluated',
            'stale' => $r['current_version_id'] === null || (int) $r['current_version_id'] !== (int) $r['version_id'] || $r['status'] !== 'ready' || $r['visibility'] !== 'public',
        ], $this->db->select('SELECT mc.n, mc.title, mc.url, mc.anchor, mc.version_id, s.current_version_id, s.status, s.visibility FROM {message_citations} mc LEFT JOIN {knowledge_sources} s ON s.id = mc.source_id AND s.site_id = mc.site_id WHERE mc.site_id = ? AND mc.message_id = ? ORDER BY mc.n', [$siteId, $messageId]));
    }

    /**
     * Inside a transaction: true if the session exists, holding its row lock until commit
     * (SELECT ... FOR UPDATE on MySQL/MariaDB; SQLite serializes write transactions).
     */
    /**
     * Creates the per-session budget bucket if it does not exist (never changes an existing one), only under the session row
     * lock (the lock an erase takes first), so an erase that lands before this point is never followed by a recreated bucket.
     * Returns false when the session no longer exists.
     */
    private function ensureSessionBucket(int $siteId, int $sessionId): bool
    {
        return $this->db->transaction(function () use ($siteId, $sessionId): bool {
            if (!$this->lockSession($siteId, $sessionId)) {
                return false;
            }
            $this->ledger->ensureBucket($siteId, 'session:' . $sessionId, $this->sessionBudgetUnits);
            return true;
        });
    }

    private function lockSession(int $siteId, int $sessionId): bool
    {
        $lock = $this->db->driver() === 'mysql' ? ' FOR UPDATE' : '';
        return $this->db->selectOne('SELECT id FROM {chat_sessions} WHERE id = ? AND site_id = ?' . $lock, [$sessionId, $siteId]) !== null;
    }

    /**
     * Appends a message (and its citations) in one transaction that first locks the session
     * row. Returns 0 if the session no longer exists, so erasure can never be undone by work
     * that was already running: SessionService::delete removes the session row first, and the
     * two serialize on that row (SELECT ... FOR UPDATE on MySQL/MariaDB; SQLite serializes
     * write transactions).
     *
     * @param list<array<string, mixed>> $citations
     */
    private function appendMessage(int $siteId, int $sessionId, int $botId, string $role, string $content, string $requestId, array $citations = [], ?string $outcome = null, ?array $groundingSourceIds = null): int
    {
        $outcome = $role === 'assistant' ? ($outcome ?? self::OUTCOME_COMPLETED) : null;
        for ($attempt = 0; $attempt < 3; $attempt++) {
            try {
                return $this->db->transaction(function () use ($siteId, $sessionId, $botId, $role, $content, $requestId, $citations, $outcome, $groundingSourceIds): int {
                    if (!$this->lockSession($siteId, $sessionId)) {
                        return 0;
                    }
                    if ($role === 'user') {
                        // Retry of the same question, so the transcript keeps it once:
                        // - after a failure that reached the provider, the unanswered copy is replaced;
                        // - after a stopped (incomplete) answer, that partial answer and its question are
                        //   replaced by the new attempt. Feedback cannot exist on a stopped answer.
                        $tail = $this->db->select('SELECT id, role, content, outcome FROM {messages} WHERE site_id = ? AND session_id = ? ORDER BY seq DESC LIMIT 2', [$siteId, $sessionId]);
                        $last = $tail[0] ?? null;
                        if ($last !== null && $last['role'] === 'user' && (string) $last['content'] === $content) {
                            $this->db->execute('DELETE FROM {messages} WHERE id = ? AND site_id = ?', [(int) $last['id'], $siteId]);
                        } elseif ($last !== null && $last['role'] === 'assistant' && $last['outcome'] === self::OUTCOME_STOPPED
                            && isset($tail[1]) && $tail[1]['role'] === 'user' && (string) $tail[1]['content'] === $content) {
                            $this->db->execute('DELETE FROM {message_citations} WHERE site_id = ? AND message_id = ?', [$siteId, (int) $last['id']]);
                            $this->db->execute('DELETE FROM {messages} WHERE id IN (?, ?) AND site_id = ?', [(int) $last['id'], (int) $tail[1]['id'], $siteId]);
                        }
                    }
                    $row = $this->db->selectOne('SELECT COALESCE(MAX(seq), 0) AS s FROM {messages} WHERE session_id = ?', [$sessionId]);
                    $id = $this->db->insert('messages', [
                        'site_id' => $siteId,
                        'session_id' => $sessionId,
                        'bot_id' => $botId,
                        'seq' => (int) ($row['s'] ?? 0) + 1,
                        'role' => $role,
                        'content' => $content,
                        'request_id' => $requestId,
                        'created_at' => $this->clock->now(),
                        'outcome' => $outcome,
                        'grounding' => $role === 'assistant' && $groundingSourceIds !== null ? ($groundingSourceIds === [] ? '' : ',' . implode(',', $groundingSourceIds) . ',') : null,
                    ]);
                    $this->storeCitations($siteId, $id, $citations);
                    return $id;
                });
            } catch (\Throwable $e) {
                if (!$this->db->isUniqueViolation($e)) {
                    throw $e;
                }
            }
        }
        // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
        throw new ChatException('conflict', 409, $requestId);
    }
}