"""Cost-controlled, single-flight Intelligence Center model generation."""
from __future__ import annotations

import hashlib
import json
import logging
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any, cast
from zoneinfo import ZoneInfo

from sqlalchemy import delete, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import AppException
from app.domains.ai_spending.service import (
    SETTING_LLM_API_KEY,
    SETTING_LLM_MODEL,
    SETTING_LLM_REGION,
)
from app.domains.bailian.llm_client import (
    DEFAULT_MODEL,
    DEFAULT_REGION,
    BailianLLMClient,
)
from app.integrations.models import PlatformSetting
from app.intelligence.contracts import (
    FLAT_KEY_SEPARATOR,
    IntelligenceContractError,
    IntelligenceOutput,
    validate_intelligence_output,
)
from app.intelligence.models import (
    IntelligenceAnalysis,
    IntelligenceGenerationLease,
    IntelligenceSnapshot,
)
from app.intelligence.privacy import UnsafeFactPackage, assert_llm_safe
from app.intelligence.scoring import canonical_json, json_value
from app.intelligence.settings import (
    DEFAULT_CLOUD_CACHE_DAYS,
    DEFAULT_CLOUD_COOLDOWN_MINUTES,
    DEFAULT_DAILY_LIMIT,
    SETTING_CLOUD_CACHE_DAYS,
    SETTING_CLOUD_COOLDOWN_MINUTES,
    SETTING_DAILY_LIMIT,
    bounded_int_setting,
    setting_value,
)

_TZ = ZoneInfo("Asia/Shanghai")
_LEASE_TTL = timedelta(minutes=5)
logger = logging.getLogger(__name__)

_SYSTEM_PROMPT = """\
You are the bilingual intelligence analyst for Helix, a company IT operating
platform. The user payload is an anonymous aggregate fact package calculated
by Helix. It is the only source of truth.

Return JSON matching the supplied schema exactly. Every narrative item must
cite scalar facts in fact_refs, and every fact_refs value must be COPIED
VERBATIM from the citable_fact_refs list in the payload. Never compose,
shorten or guess a reference; if no listed reference supports a sentence,
drop the sentence. Every number written in
Chinese or English must occur in those cited scalar facts. Use only anonymous
evidence_refs present in the package; never invent an object. Do not infer or
request names, emails, accounts, hosts, IPs, domains, credentials, policies or
other identifiers. Estimated, confirmed and realized savings are distinct.
Never describe an estimate as achieved. When insufficient_data is true, use
only low/medium confidence and include concrete data_gaps. Chinese and English
must express the same facts. Be concise and decision-useful.

Citations are examples, not inventories: at most 12 fact_refs and at most 20
evidence_refs per item, and fewer is better. The full populations are already
listed elsewhere in the product; cite the few that carry the point.

For cloud/network ownership, distinguish stored ownership from effective
ownership. A server with no direct owner field may inherit the current owner
of its cloud account/subscription. Treat owner_coverage_basis and the
directly_owned_server_count / inherited_owner_server_count / owner_gap_count
facts as authoritative. Never describe all resources without a stored owner
as ownerless, and never recommend assigning owners to inherited resources.
"""

# Subscription analyses often begin with incomplete inventory or monitoring.
# Keep the grounding boundary strict, but tell the model not to turn every
# dashboard value into an uncited numeric claim. This avoids an automatic
# second, cost-bearing request merely to recover from an avoidable rejection.
_CLOUD_SCOPE_PROMPT = _SYSTEM_PROMPT + """

For a cloud subscription, prefer a short qualitative conclusion over a
numeric recap. Do not write any numeral, percentage, date, score, count or
30/60/90 horizon unless that exact scalar is named in the item's fact_refs.
When confidence is low, return concrete data_gaps first and do not claim that
missing inventory means the subscription has no resources.
"""


# The posture summary kept being refused in production for numbers the
# model had derived CORRECTLY: handed a ratio of 0.4545 it wrote the
# obvious "45.45%", which the grounding check — a literal token
# comparison that understands no arithmetic — reported as an uncited
# number. Three rounds of publishing yet another pre-computed form
# (integer score, then each dimension, then the percentage) each bought
# one deploy before the model found a fourth way to phrase it.
#
# So the instruction is bounded instead: copy, never compute. Same
# posture as _CLOUD_SCOPE_PROMPT, which exists for this same guard.
_POSTURE_PROMPT = _SYSTEM_PROMPT + """

This is a monthly compliance posture letter for auditors. Copy numbers
verbatim from the cited scalar facts — never compute, convert, round or
re-express them. Do not turn a ratio into a percentage, do not sum or
subtract facts, and do not restate a figure in words ("about half", "two
in five"). If the form you want is not present as its own scalar fact,
describe it qualitatively instead. Name the reporting period only by
copying period_start / period_end.
"""


# The cross-domain brief was the ONLY subject whose prompt never told it
# not to do arithmetic — and it is the only subject that has never
# published. ``compliance_posture`` (the one subject with a success to its
# name) and ``cloud_scope`` each carry a numeric-discipline paragraph;
# ``platform`` inherited the bare base prompt and, once every reference
# spelling finally resolved, fell at exactly the predicted hurdle:
# "Narrative contains uncited numbers: ['2148']" — a total the model
# computed rather than copied.
#
# The guard is right to refuse it. Its package publishes 1390 numeric
# facts, so admitting "sums of two facts" would admit almost any number a
# model could write (a spurious pair adds to 2148 already); the honest fix
# is to stop the model computing, not to widen what counts as grounded.
_PLATFORM_BRIEF_PROMPT = _SYSTEM_PROMPT + """

This is a cross-domain weekly brief. Copy every figure verbatim from a
citable fact — never add, subtract, total, average, convert or round.
Do not sum counts across domains, and do not compute a share or a
percentage that is not published as its own fact. If the figure you want
does not exist as one scalar, describe the situation qualitatively
instead, or name the two published figures separately.

Keep the weekly brief deliberately short: at most 4 conclusions, 5 risks,
3 opportunities, 3 forecasts, 5 recommendations and 3 data gaps. Prefer an
empty list to a weakly supported item. This keeps every claim reviewable and
avoids a long, partially grounded response being discarded as a whole.
"""


@dataclass(frozen=True)
class GenerationResult:
    analysis: IntelligenceAnalysis
    cache_reused: bool


# A pointer the model can form must resolve to a scalar. The contract
# refuses one that lands on a dict or a list, and production proved a
# nested package makes forming a correct pointer an error-prone act rather
# than a mechanical one: the cloud subscription package nests 7 groups and
# 12 lists, and the model cited
# ``/facts/costs/data_quality/sources/2/available_count`` — concatenating
# two paths, since ``data_quality`` is a TOP-LEVEL group and not under
# ``costs``. The whole narrative was thrown away for it.
#
# So the model is handed one flat level whose keys carry the full path.
# ``costs.top_resources.0.amount`` cannot be mis-navigated: there is
# nothing to navigate. This is the same lesson the posture summary learned
# by hand, applied once at the only seam every subject passes through.
_FLAT_SEP = FLAT_KEY_SEPARATOR


def _flatten_for_citation(
    node: Any,
    prefix: str = "",
    out: dict[str, Any] | None = None,
    *,
    item_cap: int | None = None,
) -> dict[str, Any]:
    """One flat level of citable scalars, keys carrying the original path.

    List items stay individually addressable and each list also publishes
    its ``count`` — the citable fact a narrative usually wants, and the
    only one that must never be sampled away.

    ``item_cap`` bounds how many members of a list are addressed
    individually. It is not a display preference: a provider-emitted list
    scales with the roster, and the payload cost of one member is roughly
    tripled here (a long flattened key, its value, and the citable pointer
    published for it). ``count`` above is still the TRUE length, and a
    truncated list also publishes ``shown_count`` so the model can never
    read the sample as the population — see ``_FACT_LIST_ITEMS_SHOWN``.
    """
    if out is None:
        out = {}
    if isinstance(node, dict):
        for key, value in node.items():
            child = f"{prefix}{_FLAT_SEP}{key}" if prefix else str(key)
            _flatten_for_citation(value, child, out, item_cap=item_cap)
    elif isinstance(node, (list, tuple)):
        count_key = f"{prefix}{_FLAT_SEP}count" if prefix else "count"
        _assign_flat(out, count_key, len(node))
        shown = node if item_cap is None else node[:item_cap]
        if len(shown) != len(node):
            shown_key = f"{prefix}{_FLAT_SEP}shown_count" if prefix else "shown_count"
            _assign_flat(out, shown_key, len(shown))
        used: set[str] = set()
        for index, value in enumerate(shown):
            segment = _list_item_segment(value, index, prefix, used)
            used.add(segment)
            # The empty-prefix case matters: ``dimensions`` IS a list, so
            # without this branch every dimension key is published with a
            # leading separator — ``.0.metrics.1.value``. Production, 2026-08-17:
            # the model cited exactly that, character for character as it was
            # handed over, and the brief was refused for "must be a JSON
            # pointer". The address the model is given has to be an address it
            # can use. (``count`` above already had this branch; the item keys
            # did not.)
            child = f"{prefix}{_FLAT_SEP}{segment}" if prefix else segment
            _flatten_for_citation(value, child, out, item_cap=item_cap)
    else:
        _assign_flat(out, prefix or "value", node)
    return out


def _list_item_segment(
    item: Any, index: int, prefix: str, used: set[str],
) -> str:
    """Address a list item by its own name when it has one.

    A positional address is one nobody can guess. ``dimensions`` was
    published as ``0.metrics.1.value`` while every record in it carries a
    stable ``key`` (``cloud_network``, and the metric's own
    ``cloud_network.resource_waste_rate``), and production, 2026-08-17,
    showed the model writing what a human would write —
    ``cloud_network.metrics.1.value`` — against a package that only
    answered to the index. Naming the record makes the natural address the
    real one.

    Two details earn their keep. A record's key may already carry its
    parent's namespace (the metric key repeats its dimension), so the
    namespace is not published twice. And a repeated or missing key falls
    back to the index rather than colliding — two facts must never share
    one citable address (see ``_assign_flat``).
    """
    if isinstance(item, dict):
        raw = item.get("key")
        if isinstance(raw, str) and raw.strip():
            segment = raw.strip()
            head = prefix.split(_FLAT_SEP)[0] if prefix else ""
            if head and segment.startswith(f"{head}{_FLAT_SEP}"):
                segment = segment[len(head) + 1:]
            if segment and segment not in used:
                return segment
    return str(index)


def _assign_flat(out: dict[str, Any], key: str, value: Any) -> None:
    """Refuse to let two different facts share one citable key.

    Only reachable if a source key already contains a dot, which would make
    ``a.b`` ambiguous between the nested ``{"a": {"b": ...}}`` and the
    literal ``"a.b"``. Silently keeping one would hand the model a number
    filed under someone else's name, so this fails loudly in tests instead.
    """
    if key in out and out[key] != value:
        raise ValueError(
            f"Fact key collision while flattening for citation: {key!r}. "
            "A fact key must not contain a dot."
        )
    out[key] = value


# How many anonymous evidence refs of each kind the model is shown
# (2026-08-18). The platform snapshot's ``reference_map`` holds **1976**
# refs across 29 kinds (624 `access_*`, 279 `peer_*`, 148 `server_*`, …).
# Handing all of them to a model told to "cite the evidence" produced
# exactly what you would expect: single narrative items citing 42 and then
# 124 refs against a per-item cap of 20, and the whole brief — every
# correctly grounded number in it — was discarded for the count.
#
# An evidence ref anchors a claim to an example object a human can drill
# into; it is not an inventory, and the deterministic read model already
# lists every population in full. Sorted and per-kind, so one snapshot
# always shows the same sample and two runs stay comparable. Validation
# still accepts ANY ref in ``reference_map`` (``allowed_evidence_refs``),
# so narrowing what is SHOWN can never turn a legitimate citation into a
# rejection.
_EVIDENCE_REFS_PER_KIND = 6


# How many members of one provider-emitted fact list the model is shown
# individually (2026-09-01). Not a display preference — an input-window
# invariant, learned from a production outage.
#
# On 2026-08-31 seventeen transfer events landed inside the 30-day
# forecast horizon where the week before there had been one.
# ``lifecycle.transfer_forecast`` publishes a record per transfer, each
# carrying its own per-resource ``deltas`` list, and the same records are
# published again under each of the 30/60/90 horizons. That one roster
# event took ``aggregate_facts`` from 44 KB to 128 KB, and the model-facing
# package — where every leaf costs a long flattened key, its value, AND
# the citable pointer published for it — from 445 K chars to 1.05 M.
# Measured against the live endpoint: qwen3-max answered
# ``HTTP 400 InvalidParameter: Range of input length should be
# [1, 258048]``, and the weekly brief failed. The last good run was
# 114,689 input tokens for 447 K chars, i.e. ~3.9 chars/token, so 1.05 M
# chars is ~270 K tokens — over the window by 5%.
#
# Nothing about that was a one-off: every one of these lists is a record
# per person, per leaver, per transfer, per resource, and grows with the
# company. A brief narrative quotes POPULATIONS (``count``, rates, the
# dimension scores) and names a FEW examples; it has never had a use for
# the 17th transfer's 6th resource delta. So the aggregate stays exact
# (``count`` is the true length, and ``shown_count`` appears whenever a
# list is sampled) and the individual records are capped.
#
# Deterministic prefix rather than a ranked sample: the flattener is
# generic and has no way to know which field of an unknown provider's
# record means "worst". Provider order is already stable (the two large
# lists order by process / transfer id), so one snapshot always shows the
# same members and two runs stay comparable — the same property
# ``_EVIDENCE_REFS_PER_KIND`` is chosen for.
#
# 8 is sized against the real package, not chosen by taste. Rebuilt from
# the 2026-09-01 production snapshot it yields a 441,966-character prompt
# that the live endpoint answered with HTTP 200 and counted as **116,545
# input tokens** — 2.2x under the window, and within 2% of the last run
# that ever succeeded (114,689 on 2026-08-29). Every dimension score and
# every aggregate count is still exact. Note the payload does NOT fall to
# a third of the pre-fix size, because the cap applies per list and the
# nesting is multiplicative: 3 horizons x 8 transfers x 8 deltas is still
# the bulk of it. See ``test_model_fact_package_input_budget_2026_09_01``.
_FACT_LIST_ITEMS_SHOWN = 8


#: How many model replies one generation may validate before giving up
#: (2026-09-01). Sized from the measured per-attempt rate, not from taste.
#:
#: Six live runs against the real production package, with every other fix
#: in this change applied, validated **4 of 6**. The two survivors were the
#: guard working correctly — one narrative wrote ``76000`` with no fact
#: within 38,000 of it, the other wrote ``75`` for a published ``75.5``
#: (truncation is a misstatement, not an abbreviation) — so the residual is
#: single-reply slips, which is exactly what re-asking clears. At a 1/3
#: per-attempt rejection rate the job's own failure probability is
#: ``(1/3)^attempts``: one ask fails 33% of weeks, two 11%, three **3.7%**.
#:
#: Three, because the marginal ask is only ever paid on the 11% of runs
#: where two have already failed, and the thing it buys is not a nicer
#: number — it is not paging an operator with a P1 for a report that would
#: have published on the next sentence the model wrote. A fourth is where
#: this stops being insurance and starts being a retry loop against a
#: package the model genuinely cannot narrate.
#:
#: Only ``IntelligenceContractError`` is retried — a too-large or truncated
#: prompt is a property of the package and re-earns itself at full cost.
_VALIDATION_ATTEMPTS = 3


def _evidence_ref_kind(ref: str) -> str:
    """``access_017`` → ``access``. Unsuffixed refs are their own kind."""
    head, sep, tail = ref.rpartition("_")
    return head if sep and tail.isdigit() else ref


def _sampled_evidence_refs(reference_map: dict[str, Any] | None) -> list[str]:
    """Up to ``_EVIDENCE_REFS_PER_KIND`` refs per kind, deterministically."""
    per_kind: dict[str, list[str]] = {}
    for ref in sorted((reference_map or {}).keys()):
        bucket = per_kind.setdefault(_evidence_ref_kind(ref), [])
        if len(bucket) < _EVIDENCE_REFS_PER_KIND:
            bucket.append(ref)
    return sorted(ref for bucket in per_kind.values() for ref in bucket)


def _citable_pointers(package: dict[str, Any]) -> list[str]:
    """Every legal fact reference in the package, derived not hand-listed.

    The first version of this listed ``/facts/<key>`` only, and production
    immediately showed the cost: the model wanted to cite the score and
    found no legal address for it, so it constructed
    ``/facts/overall_score`` — where ``overall_score`` is a TOP-LEVEL
    member. An incomplete list of legal addresses is an invitation to
    invent one, and the score, grade and confidence are exactly what a
    narrative wants to cite.

    So the list is computed from the package itself: every scalar member,
    plus every scalar inside the already-flattened ``facts`` and
    ``dimensions``. Complete by construction, which is the only way it
    stays complete when a member is added later.

    One exclusion, and it is a caliber question rather than a completeness
    hole: an address under an ``evidence_refs`` segment holds an anonymous
    drill-down anchor (``server_79``), not a magnitude. The contract gives
    those their own channel — the per-item ``evidence_refs`` field, checked
    against the whole ``reference_map`` — and this list is what a narrative
    copies when it needs a NUMBER. Flattening turned each anchor into a
    scalar and so quietly promoted 693 of the platform package's 3505
    addresses into "facts you may cite", two thirds of everything published
    under ``dimensions``. They are still resolvable if cited, so nothing
    legal became illegal; they are simply no longer advertised as facts.
    """
    refs: list[str] = []
    for key, value in package.items():
        if isinstance(value, dict):
            refs.extend(
                f"/{key}/{child}" for child in value
                if not _is_evidence_ref_address(child)
            )
        elif not isinstance(value, list):
            refs.append(f"/{key}")
        # Lists are skipped, which also excludes every vocabulary member —
        # ``evidence_refs``, ``horizon_days`` and this list itself are all
        # lists. An explicit name-based exclusion was written first and
        # removed: a mutation showed it could never be the rule that fired,
        # so it was a branch no test could hold.
    return sorted(refs)


def _is_evidence_ref_address(flat_key: str) -> bool:
    """True for a flattened address that lands inside an ``evidence_refs`` list.

    ``cloud_network.metrics.resource_waste_rate.evidence_refs.7`` yes; its
    sibling ``.evidence_refs.count`` no — a count IS a magnitude a narrative
    may quote, and it is the fact the list-flattener publishes precisely so
    that the narrative does not have to count anchors itself.
    """
    head, sep, leaf = flat_key.rpartition(_FLAT_SEP)
    # Segment-exact: a fact legitimately named ``prior_evidence_refs`` is not
    # this, and ``endswith`` would have swallowed it.
    return (
        bool(sep)
        and head.rpartition(_FLAT_SEP)[2] == "evidence_refs"
        and leaf != "count"
    )


def build_model_fact_package(snapshot: IntelligenceSnapshot) -> dict[str, Any]:
    """Create the only payload shape permitted to cross the LLM boundary."""
    package = json_value({
        "score_version": snapshot.score_version,
        "overall_score": snapshot.overall_score,
        "confidence_score": snapshot.confidence_score,
        "insufficient_data": snapshot.confidence_score < 70,
        "grade": snapshot.grade,
        "dimensions": snapshot.dimensions,
        "facts": snapshot.aggregate_facts,
        "evidence_refs": _sampled_evidence_refs(snapshot.reference_map),
        "horizon_days": [30, 60, 90],
        "data_as_of": snapshot.data_as_of,
    }, decimal_as_string=True)
    package = cast(dict[str, Any], package)
    # Guard the NESTED shape first. ``assert_llm_safe`` matches denied KEY
    # NAMES exactly (``name``, ``email``, ``hostname``, ``ip``, ``policy``, …),
    # and flattening rewrites every key to ``<parent>.<leaf>`` — after which
    # not one of those exact names can ever match again, leaving only the value
    # regexes on the largest part of the payload. Nothing reaches the model
    # today that would trip it (the providers publish counts and bounded
    # enums), so this is defence-in-depth rather than a live leak — but this
    # function's docstring calls itself the only permitted payload shape, and a
    # future subject assembling a package here would otherwise inherit a
    # boundary with half its net missing. Checked before AND after: before for
    # the key names, after for whatever flattening and citation-listing
    # produced.
    assert_llm_safe(package)
    # ``facts`` and ``dimensions`` are the two members narratives cite;
    # ``evidence_refs`` and ``horizon_days`` stay lists because they are
    # vocabularies the contract reads, not magnitudes to quote.
    #
    # Only ``facts`` is capped. ``dimensions`` is the score model itself —
    # a closed vocabulary defined in ``catalog.py`` and held to a weight
    # total of 1.00 per dimension by ``validate_metric_weight_totals`` —
    # and it is the headline content of every brief, so a cap there would
    # silently drop a dimension the day a ninth one ships rather than
    # dropping the 9th of an unbounded population. ``facts`` is
    # provider-emitted and scales with the roster; that is the list that
    # broke the input window.
    for citable, cap in (("facts", _FACT_LIST_ITEMS_SHOWN), ("dimensions", None)):
        current = package.get(citable)
        if isinstance(current, (dict, list)):
            package[citable] = _flatten_for_citation(current, item_cap=cap)
    # The legal addresses, as a copyable list. ``evidence_refs`` has always
    # been published this way and has never once been invented; fact refs
    # were left for the model to COMPOSE from dict keys, and it composed
    # them wrong. Production evidence for the difference this makes: the
    # subject that works publishes 23 citable keys, the subject that kept
    # failing publishes 124, and its last three rejections were an
    # unresolvable path, a missing prefix, and a wholly invented
    # ``.0.metrics.0.value``. Copying a listed address cannot go wrong the
    # way composing one does.
    # Scan the real content FIRST, then attach the derived vocabulary.
    #
    # Publishing the list before the scan promoted fact KEY NAMES into
    # scanned VALUES, and dict keys were never scanned here. Real
    # production data: the cloud subscription package was rejected outright
    # with "domain-like value" on eight of its own reference strings —
    # ``/dimensions/0.score`` and friends read as hostnames. Nothing unsafe
    # had entered the payload; the same key names were already present as
    # keys of the members just validated.
    #
    # Not fixed by changing the separator: the dot is load-bearing in the
    # resolver's tolerant spellings (``FLAT_KEY_SEPARATOR`` in contracts).
    # Restoring the original values-only boundary is the smaller change and
    # the honest one.
    #
    # ``keys_are_flat_paths`` because the keys above are no longer names —
    # they are addresses this function just joined. Scanning a joined
    # address with the value regexes rejected every cloud-subscription
    # package outright: all 19 of that subject's metric keys start with
    # ``cloud.``, so the flattener emits ``...metrics.cloud.<...>`` and
    # ``metrics.cloud`` reads as ``<label>.<TLD>``. The segment scan keeps
    # the guard where content can enter and drops the reading that made our
    # own vocabulary look like hostnames; the nested pass above is what
    # actually holds the values-only boundary, and it runs unconditionally.
    assert_llm_safe(package, keys_are_flat_paths=True)
    package["citable_fact_refs"] = _citable_pointers(package)
    return package


def _utc(value: datetime) -> datetime:
    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)


def _lease_key(snapshot: IntelligenceSnapshot, analysis_kind: str) -> str:
    material = (
        f"{snapshot.subject_type}:{snapshot.subject_id}:"
        f"{snapshot.audience_key}:{analysis_kind}"
    )
    return hashlib.sha256(material.encode()).hexdigest()


async def _try_acquire_lease(lease_key: str) -> str | None:
    """Claim a visible DB lease; return None when another caller owns it."""
    from app.core.database import async_session_factory

    owner = uuid.uuid4().hex
    # A prior owner can delete its row in the tiny interval between our
    # duplicate INSERT and SELECT. Retry that benign churn instead of turning
    # a completed concurrent generation into a 500.
    for _attempt in range(3):
        now = datetime.now(UTC)
        async with async_session_factory() as session:
            inserted = False
            try:
                async with session.begin_nested():
                    session.add(IntelligenceGenerationLease(
                        lease_key=lease_key,
                        owner_token=owner,
                        expires_at=now + _LEASE_TTL,
                    ))
                    await session.flush()
                    inserted = True
            except IntegrityError:
                pass
            if inserted:
                await session.commit()
                return owner
            row = (await session.execute(
                select(IntelligenceGenerationLease)
                .where(IntelligenceGenerationLease.lease_key == lease_key)
                .with_for_update()
            )).scalar_one_or_none()
            if row is None:
                await session.rollback()
                continue
            if _utc(row.expires_at) > now:
                await session.rollback()
                return None
            row.owner_token = owner
            row.expires_at = now + _LEASE_TTL
            await session.commit()
            return owner
    # Extreme create/delete churn is equivalent to another generation owning
    # the slot; fail cheaply and let the caller poll.
    return None


async def _release_lease(lease_key: str, owner: str) -> None:
    from app.core.database import async_session_factory

    async with async_session_factory() as session:
        await session.execute(
            delete(IntelligenceGenerationLease).where(
                IntelligenceGenerationLease.lease_key == lease_key,
                IntelligenceGenerationLease.owner_token == owner,
            )
        )
        await session.commit()


async def _consume_daily_quota(user_id: uuid.UUID) -> None:
    """Debit one real model attempt in its own committed transaction."""
    from app.core.database import async_session_factory

    today = datetime.now(_TZ).date().isoformat()
    key = f"intelligence.generate_counter.{user_id}"
    async with async_session_factory() as session:
        limit = await bounded_int_setting(
            session, SETTING_DAILY_LIMIT, default=DEFAULT_DAILY_LIMIT,
            minimum=1, maximum=100,
        )
        row = await session.get(PlatformSetting, key, with_for_update=True)
        if row is None:
            try:
                async with session.begin_nested():
                    row = PlatformSetting(key=key, value={"date": today, "count": 0})
                    session.add(row)
                    await session.flush()
            except IntegrityError:
                row = await session.get(
                    PlatformSetting, key, with_for_update=True, populate_existing=True,
                )
        state = row.value if row and isinstance(row.value, dict) else {}
        used = int(state.get("count", 0)) if state.get("date") == today else 0
        if used >= limit:
            await session.rollback()
            raise AppException(
                code="INTELLIGENCE_DAILY_QUOTA",
                message=f"Daily AI analysis limit reached ({limit}/day).",
                status_code=429,
            )
        assert row is not None
        row.value = {"date": today, "count": used + 1}
        await session.commit()


async def _cached_analysis(
    db: AsyncSession,
    snapshot: IntelligenceSnapshot,
    analysis_kind: str,
    *,
    generated_after: datetime | None,
    same_facts_only: bool,
) -> IntelligenceAnalysis | None:
    conditions = [
        IntelligenceAnalysis.subject_type == snapshot.subject_type,
        IntelligenceAnalysis.subject_id == snapshot.subject_id,
        IntelligenceAnalysis.audience_key == snapshot.audience_key,
        IntelligenceAnalysis.analysis_kind == analysis_kind,
        IntelligenceAnalysis.status == "succeeded",
    ]
    if generated_after is not None:
        conditions.append(IntelligenceAnalysis.generated_at >= generated_after)
    if same_facts_only:
        conditions.append(IntelligenceAnalysis.facts_hash == snapshot.facts_hash)
    return (await db.execute(
        select(IntelligenceAnalysis)
        .where(*conditions)
        .order_by(IntelligenceAnalysis.generated_at.desc())
        .limit(1)
    )).scalar_one_or_none()


async def _reusable_analysis(
    db: AsyncSession,
    snapshot: IntelligenceSnapshot,
    analysis_kind: str,
    *,
    cloud_cache_days: int,
) -> IntelligenceAnalysis | None:
    if snapshot.subject_type == "cloud_scope":
        return await _cached_analysis(
            db, snapshot, analysis_kind,
            generated_after=datetime.now(UTC) - timedelta(days=cloud_cache_days),
            same_facts_only=True,
        )

    # Platform/domain brief: unchanged facts reuse forever. If facts changed,
    # at most one successful narrative is generated per Shanghai calendar day;
    # later changes wait for tomorrow unless a global admin explicitly forces.
    unchanged = await _cached_analysis(
        db, snapshot, analysis_kind, generated_after=None, same_facts_only=True,
    )
    if unchanged is not None:
        return unchanged
    local_midnight = datetime.now(_TZ).replace(
        hour=0, minute=0, second=0, microsecond=0,
    ).astimezone(UTC)
    return await _cached_analysis(
        db, snapshot, analysis_kind,
        generated_after=local_midnight, same_facts_only=False,
    )


async def _enforce_cooldown(
    db: AsyncSession,
    snapshot: IntelligenceSnapshot,
    analysis_kind: str,
    *,
    minutes: int,
) -> None:
    cutoff = datetime.now(UTC) - timedelta(minutes=minutes)
    recent = (await db.execute(
        select(
            IntelligenceAnalysis.status, IntelligenceAnalysis.failure_code,
        ).where(
            IntelligenceAnalysis.subject_type == snapshot.subject_type,
            IntelligenceAnalysis.subject_id == snapshot.subject_id,
            IntelligenceAnalysis.audience_key == snapshot.audience_key,
            IntelligenceAnalysis.analysis_kind == analysis_kind,
            # Every attempt counts, ``failed`` included. A failed attempt
            # already spent the tokens the provider billed for — the model
            # was called, it answered, and the answer was rejected by the
            # grounding contract or the vendor returned an error. Counting
            # only pending|succeeded let a caller hammer a subject that
            # fails deterministically (a schema drift, a bad key) with no
            # brake but the daily quota, which is per-user, not per-subject.
            IntelligenceAnalysis.status.in_(["pending", "succeeded", "failed"]),
            IntelligenceAnalysis.created_at >= cutoff,
        )
        .order_by(IntelligenceAnalysis.created_at.desc())
        .limit(1)
    )).first()
    if recent is None:
        return
    # Do not swallow the reason the last run failed: an operator who now
    # gets a throttle instead of a 502 must still be told what broke, or
    # the cooldown reads as "Helix is being difficult". Only the stable
    # failure CODE travels here — ``failure_message`` is model-adjacent
    # and stays behind the privileged projection.
    reason = (
        f" The last attempt failed with {recent.failure_code}."
        if recent.status == "failed" and recent.failure_code
        else ""
    )
    raise AppException(
        code="INTELLIGENCE_COOLDOWN",
        message=f"AI analysis is limited to once every {minutes} minutes.{reason}",
        status_code=429,
    )


async def latest_analysis(
    db: AsyncSession,
    *,
    subject_type: str,
    subject_id: str,
    audience_key: str,
    analysis_kind: str,
) -> IntelligenceAnalysis | None:
    """Read only. This function can never invoke a model."""
    return (await db.execute(
        select(IntelligenceAnalysis).where(
            IntelligenceAnalysis.subject_type == subject_type,
            IntelligenceAnalysis.subject_id == subject_id,
            IntelligenceAnalysis.audience_key == audience_key,
            IntelligenceAnalysis.analysis_kind == analysis_kind,
            IntelligenceAnalysis.status == "succeeded",
        ).order_by(IntelligenceAnalysis.generated_at.desc()).limit(1)
    )).scalar_one_or_none()


def _persist_output(row: IntelligenceAnalysis, output: IntelligenceOutput) -> None:
    row.summary_zh = output.summary_zh
    row.summary_en = output.summary_en
    row.summary_fact_refs = output.summary_fact_refs
    row.conclusions = [item.model_dump(mode="json") for item in output.conclusions]
    row.risks = [item.model_dump(mode="json") for item in output.risks]
    row.opportunities = [item.model_dump(mode="json") for item in output.opportunities]
    row.forecasts = [item.model_dump(mode="json") for item in output.forecasts]
    row.recommendations = [item.model_dump(mode="json") for item in output.recommendations]
    row.data_gaps = [item.model_dump(mode="json") for item in output.data_gaps]


async def _mark_failed(
    db: AsyncSession,
    row_id: uuid.UUID,
    *,
    code: str,
    message: str,
    model_used: str | None = None,
    input_tokens: int | None = None,
    output_tokens: int | None = None,
) -> None:
    """Recover a possibly failed transaction and retain a bounded failure row.

    Best-effort by construction, for the same reason
    ``_analysis_failure_reason`` is: this runs on a path that is ALREADY
    failing, and the caller's next statement re-raises the real cause. If
    persisting the audited row is allowed to raise, that raise replaces the
    diagnosis with whatever the audit write tripped over — the caller's
    ``raise AppException(...) from exc`` never executes, and both the
    operator alert and the ``scheduled_job_runs`` row name the bookkeeping
    fault instead of what actually broke.

    Measured, not theorised (2026-09-06). Two real faults, injected without
    mocking an exception:

    * a pooled connection dropped while the model was answering — the
      documented cost of holding one across this job's 120s LLM call, which
      is why the handler is registered ``weight="heavy"`` — surfaced as
      ``AI 大脑每日跨域简报生成失败:7s2a``;
    * a vendor ``_meta.model_used`` wider than the 128-char column (see the
      truncation below) surfaced as ``…失败:9h9h``, discarding a
      ``Narrative contains uncited numbers: ['9999']`` that all three
      validation attempts had already agreed on.

    ``7s2a`` and ``9h9h`` are SQLAlchemy DOC-ANCHOR slugs
    (``PendingRollbackError``, ``DataError``) that reach the alert because
    every DBAPI error carries a ``.code`` attribute; see the handler, which
    no longer trusts one. In both runs the audited row was lost as well, so
    the prod-TiDB query that every previous incident here was diagnosed by
    would have returned nothing.
    """
    try:
        await db.rollback()
        row = await db.get(IntelligenceAnalysis, row_id, populate_existing=True)
        if row is None:
            return
        row.status = "failed"
        row.failure_code = code[:64]
        row.failure_message = message[:512]
        # Bounded like its two neighbours, and for a sharper reason: ``code``
        # and ``message`` are ours, while this value is the VENDOR's —
        # ``_meta.model_used`` straight off the DashScope response — written
        # into ``String(128)``. Leaving one caller-supplied string unbounded
        # next to two truncated ones is what turned a diagnosed rejection
        # into an undiagnosable DataError.
        row.model_used = model_used[:128] if model_used else model_used
        row.input_token_count = input_tokens
        row.output_token_count = output_tokens
        row.generated_at = datetime.now(UTC)
        await db.commit()
    except Exception:  # noqa: BLE001 — never replace the caller's real cause
        logger.exception(
            "intelligence: could not persist the audited failure row for "
            "analysis %s (code %s); the original cause is re-raised by the "
            "caller", row_id, code,
        )
        # Hand the session back usable. A failed flush deactivates the whole
        # transaction, so swallowing without rolling back just moves the
        # ``PendingRollbackError`` to whoever touches the session next — the
        # exact shape ``check_savepoint_isolation`` exists to stop, and the
        # reason a swallow alone would still have cost us the diagnosis.
        try:
            await db.rollback()
        except Exception:  # noqa: BLE001 — the connection may simply be gone
            logger.warning(
                "intelligence: session unusable after a failed audit write "
                "for analysis %s", row_id, exc_info=True,
            )


# Output budget for one analysis (2026-08-17). The daily cross-domain
# brief is the widest contract Helix asks for — bilingual narrative plus
# up to 12 conclusions, 20 risks, 20 opportunities, 20 recommendations
# and 20 data gaps, each with a zh+en title and detail. At the previous
# 6000-token ceiling the model was cut off mid-object (production runs
# recorded output_token_count 5994 and 5439), and a truncated JSON
# document is an invalid one: the subject failed with
# ``BAILIAN_LLM_BAD_JSON`` and has never once published.
#
# 16000 leaves ~2.7x headroom over the observed truncation point, and
# the citation cap moving 20 → 40 makes replies longer, not shorter, so
# sizing this to today's output would just buy one deploy. It is a
# ceiling, not a spend — billing follows the tokens actually emitted,
# and the subjects that do publish land under 2000. Probed against the
# live DashScope endpoint on 2026-08-17: qwen3-max returns HTTP 200 for
# max_tokens up to 65536, so this stays well inside the model's limit.
DEFAULT_MAX_OUTPUT_TOKENS = 16000


@dataclass(frozen=True, slots=True)
class GenerationPolicy:
    """Per-subject cache, cooldown, prompt and output-contract behaviour."""

    cache_days_setting: str | None = None
    cooldown_setting: str | None = None
    default_cache_days: int = 1
    default_cooldown_minutes: int = 1
    output_model: type[IntelligenceOutput] = IntelligenceOutput
    system_prompt: str | None = None
    # Per-subject override for the model's output budget. A subject with
    # a narrower contract can lower it; nothing today needs to.
    max_output_tokens: int = DEFAULT_MAX_OUTPUT_TOKENS


_DEFAULT_POLICY = GenerationPolicy()
_POLICIES: dict[str, GenerationPolicy] = {
    # Monthly compliance posture: a settled month never changes, so the
    # cache window is long and the cooldown only guards a double-click.
    # Registered here rather than at the auditor domain's import time so
    # the whole table stays readable in one place.
    "compliance_posture": GenerationPolicy(
        default_cache_days=30,
        default_cooldown_minutes=5,
        system_prompt=_POSTURE_PROMPT,
    ),
    # The weekly cross-domain brief. Same cache/cooldown as the default;
    # registered solely to give it the numeric discipline every other
    # subject already had.
    "platform": GenerationPolicy(system_prompt=_PLATFORM_BRIEF_PROMPT),
    "cloud_scope": GenerationPolicy(
        cache_days_setting=SETTING_CLOUD_CACHE_DAYS,
        cooldown_setting=SETTING_CLOUD_COOLDOWN_MINUTES,
        default_cache_days=DEFAULT_CLOUD_CACHE_DAYS,
        default_cooldown_minutes=DEFAULT_CLOUD_COOLDOWN_MINUTES,
        system_prompt=_CLOUD_SCOPE_PROMPT,
    ),
}


def register_generation_policy(subject_type: str, policy: GenerationPolicy) -> None:
    """Register explicitly owned model-generation behaviour for a subject."""
    if not subject_type.strip():
        raise ValueError("GenerationPolicy subject_type must not be blank")
    if not 1 <= policy.default_cache_days <= 30:
        raise ValueError("GenerationPolicy.default_cache_days must be within 1..30")
    if not 1 <= policy.default_cooldown_minutes <= 1440:
        raise ValueError("GenerationPolicy.default_cooldown_minutes must be within 1..1440")
    if not 1000 <= policy.max_output_tokens <= 32000:
        raise ValueError("GenerationPolicy.max_output_tokens must be within 1000..32000")
    if not isinstance(policy.output_model, type) or not issubclass(
        policy.output_model, IntelligenceOutput,
    ):
        raise ValueError("GenerationPolicy.output_model must inherit IntelligenceOutput")
    _POLICIES[subject_type] = policy


def policy_for(subject_type: str) -> GenerationPolicy:
    """Return the named policy; unknown subjects use the visible default."""
    return _POLICIES.get(subject_type, _DEFAULT_POLICY)


async def generate_analysis(
    db: AsyncSession,
    *,
    snapshot: IntelligenceSnapshot,
    analysis_kind: str,
    actor_id: uuid.UUID | None,
    force: bool = False,
    enforce_user_quota: bool = True,
) -> GenerationResult:
    """Generate one audited analysis, or return an unchanged-facts cache hit."""
    if snapshot.data_as_of is None:
        raise AppException(
            code="INTELLIGENCE_NO_DATA",
            message="Analysis requires at least one dated deterministic fact source.",
            status_code=409,
        )
    policy = policy_for(snapshot.subject_type)
    cache_days = (
        await bounded_int_setting(
            db, policy.cache_days_setting,
            default=policy.default_cache_days, minimum=1, maximum=30,
        )
        if policy.cache_days_setting
        else policy.default_cache_days
    )
    if not force:
        cached = await _reusable_analysis(
            db, snapshot, analysis_kind, cloud_cache_days=cache_days,
        )
        if cached is not None:
            # Persist the fact on the row that is being served. The column
            # existed and was read back by ``read_latest_analysis`` and the
            # subscription projection, but nothing ever wrote True — so the
            # brief reported ``cache_reused: false`` for a narrative that
            # had not been regenerated for days, and the only truthful
            # signal was the in-request ``GenerationResult`` flag that a
            # later GET cannot see. Monotonic: "this stored analysis has
            # been served from cache at least once", never reset.
            if not cached.cache_reused:
                cached.cache_reused = True
                await db.commit()
            return GenerationResult(analysis=cached, cache_reused=True)

    lease_key = _lease_key(snapshot, analysis_kind)
    owner = await _try_acquire_lease(lease_key)
    if owner is None:
        raise AppException(
            code="INTELLIGENCE_GENERATION_IN_PROGRESS",
            message="An analysis for this subject is already being generated.",
            status_code=409,
        )
    try:
        cooldown = (
            await bounded_int_setting(
                db, policy.cooldown_setting,
                default=policy.default_cooldown_minutes,
                minimum=1, maximum=1440,
            )
            if policy.cooldown_setting
            else policy.default_cooldown_minutes
        )
        if not force:
            await _enforce_cooldown(
                db, snapshot, analysis_kind, minutes=cooldown,
            )
        if enforce_user_quota:
            if actor_id is None:
                raise AppException(
                    code="INTELLIGENCE_ACTOR_REQUIRED",
                    message="User-triggered analysis requires an actor.", status_code=400,
                )

        row = IntelligenceAnalysis(
            snapshot_id=snapshot.id,
            subject_type=snapshot.subject_type,
            subject_id=snapshot.subject_id,
            audience_key=snapshot.audience_key,
            analysis_kind=analysis_kind,
            facts_hash=snapshot.facts_hash,
            generation_key=hashlib.sha256(f"{snapshot.id}:{owner}".encode()).hexdigest(),
            status="pending",
            generated_by_user_id=actor_id,
            data_as_of=snapshot.data_as_of,
            forced=force,
        )
        db.add(row)
        await db.commit()  # pending is visible to concurrent GET/polling
        await db.refresh(row)

        # Everything the FAILURE path needs, held outside the ORM from here
        # on. The same lesson as the 2026-08-17 ``MissingGreenlet`` incident
        # one layer up (see the handler's ``snapshot_id = str(snapshot.id)``):
        # once generation has gone wrong, the session may be rolled back or
        # deactivated, and a sync attribute read on an instance in that
        # session emits a refresh SELECT instead of returning a value.
        #
        # The except branches below used to pass ``row.model_used`` and the
        # two token counts straight as ARGUMENTS to ``_mark_failed``. Python
        # evaluates arguments before the call, so on 2026-09-06 a genuinely
        # dropped connection made that read raise ``PendingRollbackError``
        # BEFORE ``_mark_failed`` was entered — no audited row was written at
        # all, and the alert named the read rather than the fault. Mirroring
        # each value into a local costs three assignments and means the
        # failure path never touches the session it is recovering from.
        row_id = row.id
        used_model: str | None = None
        used_input_tokens: int | None = None
        used_output_tokens: int | None = None

        try:
            facts = build_model_fact_package(snapshot)
            api_key = await setting_value(db, SETTING_LLM_API_KEY)
            if not isinstance(api_key, str) or not api_key.strip():
                raise AppException(
                    code="BAILIAN_LLM_NO_KEY",
                    message="No Bailian API key is configured.", status_code=503,
                )
            model_value = await setting_value(db, SETTING_LLM_MODEL)
            region_value = await setting_value(db, SETTING_LLM_REGION)
            model = (
                model_value
                if isinstance(model_value, str) and model_value.strip()
                else DEFAULT_MODEL
            )
            region = (
                region_value
                if isinstance(region_value, str) and region_value.strip()
                else DEFAULT_REGION
            )
            if enforce_user_quota:
                assert actor_id is not None  # validated before pending row creation
                await _consume_daily_quota(actor_id)
            client = BailianLLMClient(api_key, model=model, region=region, timeout=120.0, purpose="intelligence.analysis")
            output_model = policy.output_model
            system_prompt = (
                (policy.system_prompt or _SYSTEM_PROMPT) + "\nJSON Schema:\n"
                + json.dumps(output_model.model_json_schema(), ensure_ascii=False)
            )
            user_payload = canonical_json(facts)
            allowed_refs = set((snapshot.reference_map or {}).keys())
            # A grounding or schema rejection is a property of THIS reply, not
            # of the facts: the same package re-asked produces different
            # wording, and the guard is deliberately strict enough that a
            # compliant narrator still trips it sometimes. Measured on the
            # 2026-09-01 production package, six live runs: one reply cited a
            # metric that does not exist and one wrote a ratio it had been
            # told not to compute — both are single-reply slips, and both
            # cost the weekly job a P1 alert for a report that would have
            # published on a second ask.
            #
            # Only this class is retried. ``BAILIAN_LLM_INPUT_TOO_LARGE`` and
            # ``BAILIAN_LLM_OUTPUT_TRUNCATED`` are properties of the PACKAGE
            # and a retry re-earns them at the same cost, so they stay out
            # (that is why the loop catches the contract error specifically
            # rather than everything).
            last_error: IntelligenceContractError | None = None
            output = None
            for attempt in range(_VALIDATION_ATTEMPTS):
                raw = await client.chat_json(
                    system=system_prompt,
                    user=user_payload,
                    max_tokens=policy.max_output_tokens,
                    temperature=0.1,
                )
                raw_meta = raw.get("_meta")
                meta: dict[str, Any] = raw_meta if isinstance(raw_meta, dict) else {}
                # Bounded to the column here too, so an over-wide vendor id
                # cannot fail the SUCCESS commit either — a run whose
                # narrative validated has earned publication, and the model
                # id is a label on it, not a reason to discard it.
                used_model = str(meta.get("model_used") or model)[:128]
                # Tokens ACCUMULATE across attempts. Every call is billed, and
                # a row reporting only the last attempt would understate the
                # spend of exactly the runs that cost the most.
                used_input_tokens = (
                    (used_input_tokens or 0) + (meta.get("input_tokens") or 0)
                ) or None
                used_output_tokens = (
                    (used_output_tokens or 0) + (meta.get("output_tokens") or 0)
                ) or None
                row.model_used = used_model
                row.input_token_count = used_input_tokens
                row.output_token_count = used_output_tokens
                try:
                    output = validate_intelligence_output(
                        raw,
                        facts=facts,
                        allowed_evidence_refs=allowed_refs,
                        confidence_score=snapshot.confidence_score,
                        output_model=output_model,
                    )
                    break
                except IntelligenceContractError as contract_error:
                    last_error = contract_error
                    logger.warning(
                        "intelligence %s attempt %d/%d failed validation: %s",
                        analysis_kind, attempt + 1, _VALIDATION_ATTEMPTS,
                        contract_error,
                    )
            if output is None:
                # Every attempt was rejected. Raise the LAST reason so the
                # failure row and the operator alert name a real rule, which
                # is what ``_analysis_failure_reason`` reads.
                assert last_error is not None
                raise last_error
            _persist_output(row, output)
            row.status = "succeeded"
            row.generated_at = datetime.now(UTC)
            await db.commit()
            await db.refresh(row)
            return GenerationResult(analysis=row, cache_reused=False)
        except UnsafeFactPackage as exc:
            await _mark_failed(
                db, row_id, code="INTELLIGENCE_FACTS_UNSAFE",
                message="Aggregate facts failed the model privacy boundary.",
            )
            raise AppException(
                code="INTELLIGENCE_FACTS_UNSAFE",
                message="Analysis facts did not pass the privacy boundary.",
                status_code=500,
            ) from exc
        except IntelligenceContractError as exc:
            await _mark_failed(
                db, row_id, code="INTELLIGENCE_OUTPUT_INVALID",
                # Name the rule that rejected the output. Without it every
                # rejection reads identically and an operator cannot tell an
                # invented number from a schema drift without re-running the
                # model. ``_mark_failed`` truncates to the column width, and
                # the narrative is already past the privacy boundary — the
                # facts reaching the model are aggregate-only.
                message=(
                    "Model output failed grounding or schema validation: "
                    f"{exc}"
                ),
                model_used=used_model,
                input_tokens=used_input_tokens,
                output_tokens=used_output_tokens,
            )
            raise AppException(
                code="INTELLIGENCE_OUTPUT_INVALID",
                message="AI output could not be verified against Helix facts.",
                status_code=502,
            ) from exc
        except AppException as exc:
            # Carry the provider's own sentence, not just the code. The
            # code alone ("Generation request failed with
            # BAILIAN_LLM_BAD_JSON.") is what every operator saw for
            # every daily-brief failure: it names the symptom and hides
            # the cause — the token budget, the HTTP status, the
            # rate-limit reason — which cost a live-endpoint probe to
            # recover. Scrubbed because vendor error bodies occasionally
            # echo credential-shaped fragments; ``_mark_failed``
            # truncates to the column width.
            from app.integrations.service import _scrub_message  # noqa: PLC0415
            try:
                detail = _scrub_message(str(exc) or "")
            except Exception:  # noqa: BLE001 — never fail the failure path
                detail = ""
            detail = detail.strip().split("\n", 1)[0]
            await _mark_failed(
                db, row_id, code=exc.code,
                message=(
                    f"Generation request failed with {exc.code}: {detail}"
                    if detail else f"Generation request failed with {exc.code}."
                ),
            )
            raise
        except Exception as exc:
            # Name the fault, the way the ``AppException`` branch above names
            # the vendor's. This is the branch every non-model failure lands
            # in — a dropped pool connection across the 120s call, a deadlock,
            # a serialization error — and "Unexpected model generation
            # failure." described none of them. It is also the one branch
            # where nothing else records the cause: there is no vendor code
            # and no contract rule, so if this row does not carry the
            # exception, the run's only description of what broke is a
            # generic sentence and an operator has to reproduce it. Scrubbed
            # and clipped for the same reason as above — these strings can
            # echo credential-shaped fragments, and ``_mark_failed`` truncates
            # to the column width in any case.
            try:
                from app.integrations.service import _scrub_message  # noqa: PLC0415
                fault = _scrub_message(str(exc) or "").strip().split("\n", 1)[0]
            except Exception:  # noqa: BLE001 — never fail the failure path
                fault = ""
            await _mark_failed(
                db, row_id, code="INTELLIGENCE_GENERATION_FAILED",
                message=(
                    f"Unexpected model generation failure: "
                    f"{type(exc).__name__}: {fault}"
                    if fault else
                    f"Unexpected model generation failure: {type(exc).__name__}"
                ),
                model_used=used_model,
                input_tokens=used_input_tokens,
                output_tokens=used_output_tokens,
            )
            raise AppException(
                code="INTELLIGENCE_GENERATION_FAILED",
                message="AI analysis generation failed.", status_code=502,
            ) from exc
    finally:
        try:
            await _release_lease(lease_key, owner)
        except Exception:  # noqa: BLE001 — lease expires; never mask real result
            logger.exception("intelligence_generation_lease_release_failed")