"""Reading the whole corpus and saying what is wrong with it.

The knowledge-gap digest that ships today ends at "these questions got zero
hits". That is a true statement and an unusable one: it tells IT the corpus has
holes without saying which page to touch, so the work it creates is *starting an
investigation*, every week, forever. Nothing in it can be clicked.

Everything here ends in a proposal instead — a specific change to a specific
page, with its reasoning and its evidence, that a person accepts or rejects.

Four passes, each fixing a defect the corpus is known to have:

* **duplicates and contradictions** — nothing has ever compared two pages to
  each other, so a corpus assembled from four sources (hand-written, drafted
  from tickets, imported from the service desk, mirrored from Feishu) has never
  been checked for saying the same thing twice or two different things once;
* **missing vocabulary** — the measured recall ceiling. Questions that hit
  nothing are matched back against the pages that *should* have answered them,
  and the output is the word to add;
* **dead links** — an edge pointing at a page that has since been archived,
  which is exactly what the Feishu mirror's disappearance sweep produces;
* **pages with no summary** — the index degrades to body-heads, which is the
  difference between a list you can reason about and one you cannot.

**Why the model is asked to confirm, and never to decide alone.** Cosine
similarity finds *pairs worth looking at* and cannot tell "the same thing" from
"the opposite of the same thing" — 「怎么连 VPN」 and 「VPN 连不上」 sit very
close together and are two different pages. So the pass proposes nothing on
similarity alone; the model classifies the pair, and a person decides. Both
gates matter: without the model the queue fills with near-misses, without the
person a bad classification quietly merges away a page.
"""
from __future__ import annotations

import json
import logging
import math
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.knowledge_loop import service as loop
from app.core.knowledge_loop.constants import (
    CORPUS_SUPPORT_KB,
    KIND_ADD_ALIASES,
    KIND_CONTRADICTION,
    KIND_DEAD_LINK,
    KIND_LINK,
    KIND_MERGE,
    KIND_NEW_ARTICLE,
    SOURCE_ANSWER_BACKFILL,
    SOURCE_EVAL_FAILURE,
    SOURCE_LINT,
)
from app.domains.support.kb_wiki import fallback_summary, index
from app.domains.support.models import (
    KB_ARCHIVED,
    KB_PUBLISHED,
    KB_SOURCE_FEISHU_DOC,
    SupportKbArticle,
    SupportKbArticleLink,
)

logger = logging.getLogger(__name__)

#: Cosine above which two published pages are worth a second look.
#:
#: **This number is not measured yet** and the code is arranged so that it does
#: not have to be: it only decides which pairs are shown to the classifier, and
#: the classifier plus a human decide everything after that. Too low costs model
#: calls; too high misses duplicates. The first real run's accept/reject ratio
#: is what should move it, and until then it is a starting point, not a finding.
PAIR_SIMILARITY_FLOOR = 0.86

#: Pairs classified per pass. A corpus of 250 pages has ~31,000 pairs; the floor
#: cuts that to a handful, but a corpus that grows or an embedding model that
#: shifts could produce hundreds, and each one is a model call.
MAX_PAIRS_PER_PASS = 25

#: Gap queries considered per pass.
MAX_GAP_QUERIES = 40

#: A gap query is only turned into an alias suggestion when the corpus has a
#: page that plausibly should have answered it. Below this there is no page to
#: attach the word to — that is a missing *document*, which the digest already
#: reports and which nothing here can invent.
GAP_ATTACH_FLOOR = 0.55


@dataclass
class LintReport:
    duplicates: int = 0
    contradictions: int = 0
    links: int = 0
    aliases: int = 0
    dead_links: int = 0
    pairs_examined: int = 0
    gaps_examined: int = 0
    duplicates_skipped: int = 0
    llm_available: bool = True

    def as_dict(self) -> dict[str, Any]:
        return {
            "duplicates": self.duplicates,
            "contradictions": self.contradictions,
            "links": self.links,
            "aliases": self.aliases,
            "dead_links": self.dead_links,
            "pairs_examined": self.pairs_examined,
            "gaps_examined": self.gaps_examined,
            "duplicates_skipped": self.duplicates_skipped,
            "llm_available": self.llm_available,
            "proposals": (
                self.duplicates
                + self.contradictions
                + self.links
                + self.aliases
                + self.dead_links
            ),
        }


def _cosine(a: list[float], b: list[float]) -> float:
    if not a or not b or len(a) != len(b):
        return 0.0
    dot = sum(x * y for x, y in zip(a, b, strict=False))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(x * x for x in b))
    if na == 0.0 or nb == 0.0:
        return 0.0
    return dot / (na * nb)


# ---------------------------------------------------------------------------
# Pass 1 — pairs: duplicate / contradiction / related
# ---------------------------------------------------------------------------

_PAIR_SYSTEM = """你在给一个企业 IT 知识库做去重与冲突排查。给定两条已发布条目
(各含标题、一句话说明、正文节选),判断它们的关系。只能选一种:

- "duplicate":讲的是同一件事,信息基本重合,应该合并成一条。
- "contradicts":讲的是同一件事,但说法互相冲突(口径、年限、流程不一致),
  不能合并 —— 需要有人去判断哪个是现行的。
- "related":有关联但讲的是不同的事(比如一个是配置、一个是排障),应该互相
  链接,但不该合并。
- "unrelated":只是措辞像,实际没关系。

**默认答案是 "unrelated"。** 这两条是靠向量相似度挑出来的,而向量分不清
「同一件事」和「同一件事的反面」——「怎么连 VPN」和「VPN 连不上怎么办」相似度
很高,但它们是两条不同的条目。只有在你能说出**同一件事**是哪件事的时候,才选
前三种。

选 "duplicate" 时给出 keep_id(内容更完整、更新的那一条)。reason 用中文一句话
说清依据。"""


async def _published_vectors(db: AsyncSession) -> dict[uuid.UUID, list[float]]:
    rows = (
        await db.execute(
            select(SupportKbArticle.id, SupportKbArticle.embedding).where(
                SupportKbArticle.status == KB_PUBLISHED,
                SupportKbArticle.embedding.is_not(None),
                # Mirrored pages cannot be edited or merged here, so proposing
                # anything about them would produce a queue item whose only
                # possible outcome is a refusal at apply time.
                SupportKbArticle.source != KB_SOURCE_FEISHU_DOC,
            )
        )
    ).all()
    return {row[0]: row[1] for row in rows if isinstance(row[1], list) and row[1]}


async def _existing_pair_links(
    db: AsyncSession,
) -> set[tuple[uuid.UUID, uuid.UUID]]:
    """Pairs that already carry an edge — no need to propose one again."""
    rows = (
        await db.execute(
            select(
                SupportKbArticleLink.from_article_id,
                SupportKbArticleLink.to_article_id,
            )
        )
    ).all()
    out: set[tuple[uuid.UUID, uuid.UUID]] = set()
    for left, right in rows:
        out.add((left, right))
        out.add((right, left))
    return out


async def lint_pairs(db: AsyncSession, *, report: LintReport) -> None:
    vectors = await _published_vectors(db)
    if len(vectors) < 2:
        return
    linked = await _existing_pair_links(db)

    ids = sorted(vectors, key=str)
    scored: list[tuple[float, uuid.UUID, uuid.UUID]] = []
    for position, first_id in enumerate(ids):
        for second_id in ids[position + 1:]:
            if (first_id, second_id) in linked:
                continue
            score = _cosine(vectors[first_id], vectors[second_id])
            if score >= PAIR_SIMILARITY_FLOOR:
                scored.append((score, first_id, second_id))
    scored.sort(key=lambda item: (-item[0], str(item[1]), str(item[2])))
    shortlist = scored[:MAX_PAIRS_PER_PASS]
    report.pairs_examined = len(shortlist)
    if not shortlist:
        return

    wanted = {item for _s, one, two in shortlist for item in (one, two)}
    articles = {
        article.id: article
        for article in (
            (
                await db.execute(
                    select(SupportKbArticle).where(SupportKbArticle.id.in_(wanted))
                )
            )
            .scalars()
            .all()
        )
    }

    from app.domains.support.copilot import one_shot_json  # noqa: PLC0415

    for score, left_id, right_id in shortlist:
        maybe_left = articles.get(left_id)
        maybe_right = articles.get(right_id)
        if maybe_left is None or maybe_right is None:
            continue
        left, right = maybe_left, maybe_right
        verdict = await one_shot_json(
            db,
            _PAIR_SYSTEM,
            json.dumps(
                {
                    "a": _pair_entry(left),
                    "b": _pair_entry(right),
                    "cosine": round(score, 3),
                },
                ensure_ascii=False,
            )
            + '\n\n请输出 JSON:{"relation":"duplicate|contradicts|related|'
            'unrelated","keep_id":"重复时保留的那一条 id 或 null",'
            '"reason":"中文一句话"}',
            timeout=45.0,
            max_tokens=600,
        )
        if verdict is None:
            report.llm_available = False
            # No classifier means every remaining pair would be guessed at.
            # Stopping is better than filling the queue with cosine scores.
            return
        relation = str(verdict.get("relation") or "").strip().lower()
        reason = str(verdict.get("reason") or "").strip()[:1000] or None
        if relation == "unrelated":
            continue

        if relation == "duplicate":
            keep_raw = str(verdict.get("keep_id") or "").strip()
            keep, drop = (
                (left, right) if keep_raw == str(left.id) else (right, left)
            )
            if keep_raw not in (str(left.id), str(right.id)):
                # It said duplicate but not which to keep. Newer wins — and the
                # reviewer sees both bodies side by side anyway.
                keep, drop = (
                    (left, right)
                    if (left.updated_at or left.created_at)
                    >= (right.updated_at or right.created_at)
                    else (right, left)
                )
            if KB_SOURCE_FEISHU_DOC in (drop.source, keep.source):
                # Cannot write a mirrored page from here; propose a link so the
                # pair is still visible together.
                #
                # BOTH sides, not just the archived one. A merge writes the
                # SURVIVOR too — its aliases gain the absorbed page's and its
                # embedding is cleared — so a merge INTO a mirror is refused at
                # apply time just the same. Checking only ``drop`` produced a
                # row whose only outcomes were reject and accept-and-error, and
                # (since a failure now clears the fingerprint) one the next
                # weekly lint would file again.
                report.duplicates_skipped += 1
                relation = "related"
            else:
                row = await loop.propose(
                    db,
                    corpus=CORPUS_SUPPORT_KB,
                    kind=KIND_MERGE,
                    title=f"合并「{drop.title}」到「{keep.title}」",
                    source=SOURCE_LINT,
                    target_ref=str(keep.id),
                    related_ref=str(drop.id),
                    # ⚠️ EMPTY on purpose. ``fingerprint_for`` hashes the
                    # payload, and the score is not part of what this proposal
                    # ASKS FOR — it is why it was raised, which is what
                    # ``evidence`` is for and why the fingerprint excludes it.
                    # Carrying the number here defeated the dedup entirely:
                    # accepting an alias proposal on either page clears its
                    # embedding, the re-embed shifts the cosine by a thousandth,
                    # and the next run files a SECOND open proposal for the same
                    # pair — one row per re-embed, forever.
                    payload={},
                    rationale=reason,
                    evidence={"cosine": round(score, 3)},
                )
                if row:
                    report.duplicates += 1
                continue

        kind = KIND_CONTRADICTION if relation == "contradicts" else KIND_LINK
        row = await loop.propose(
            db,
            corpus=CORPUS_SUPPORT_KB,
            kind=kind,
            title=(
                f"「{left.title}」与「{right.title}」说法冲突"
                if kind == KIND_CONTRADICTION
                else f"关联「{left.title}」与「{right.title}」"
            ),
            source=SOURCE_LINT,
            target_ref=str(left.id),
            related_ref=str(right.id),
            payload={},
            rationale=reason,
            evidence={"cosine": round(score, 3)},
        )
        if row:
            if kind == KIND_CONTRADICTION:
                report.contradictions += 1
            else:
                report.links += 1


def _pair_entry(article: SupportKbArticle) -> dict[str, Any]:
    return {
        "id": str(article.id),
        "title": article.title,
        "summary": article.summary or fallback_summary(article),
        "body": (article.body or "")[:1500],
        "updated_at": str(article.updated_at or ""),
    }


# ---------------------------------------------------------------------------
# Pass 2 — vocabulary
# ---------------------------------------------------------------------------

_ALIAS_SYSTEM = """一个员工这样问,而知识库没有召回到对的条目。给定他的原话和
知识库里**最应该回答这个问题**的那条条目,判断:

- 如果那条条目**确实**回答了这个问题,只是措辞对不上 —— 给出应该补进条目别名的
  说法(用员工的词,不是条目里的词),relation 填 "vocabulary"。
- 如果那条条目**并没有**回答这个问题 —— relation 填 "not_covered",别名留空。
  这是内容缺口,不是词表问题,补别名会把不相关的问题引到错的条目上,比不召回更糟。

reason 用中文一句话。"""


async def lint_vocabulary(
    db: AsyncSession, *, report: LintReport, window_days: int = 14,
) -> None:
    """Turn questions that hit nothing into the word that was missing."""
    from app.domains.support.copilot import one_shot_json  # noqa: PLC0415
    from app.domains.support.kb_service import SupportKbService  # noqa: PLC0415
    from app.domains.support.models import (  # noqa: PLC0415
        DEFLECT_ESCALATED,
        DEFLECT_EXPIRED,
        SupportDeflectionEvent,
    )

    since = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=window_days)
    rows = (
        (
            await db.execute(
                select(SupportDeflectionEvent.query)
                .where(
                    SupportDeflectionEvent.created_at >= since,
                    SupportDeflectionEvent.outcome.in_(
                        (DEFLECT_ESCALATED, DEFLECT_EXPIRED),
                    ),
                )
                .order_by(SupportDeflectionEvent.created_at.desc())
                .limit(200)
            )
        )
        .scalars()
        .all()
    )
    seen: set[str] = set()
    queries: list[str] = []
    for raw in rows:
        cleaned = " ".join((raw or "").split())
        key = cleaned.casefold()
        if len(cleaned) < 4 or key in seen:
            continue
        seen.add(key)
        queries.append(cleaned)
        if len(queries) >= MAX_GAP_QUERIES:
            break
    report.gaps_examined = len(queries)
    if not queries:
        return

    service = SupportKbService(db)
    for question in queries:
        # The page that SHOULD have answered it. Semantic recall is used with
        # its own floor here rather than the employee-facing one: the whole
        # point is that the employee-facing path found nothing, so this asks a
        # looser question — "is there a page about this at all".
        found = await service.search_published(question, limit=1)
        if not found:
            continue
        article = found[0]
        if article.source == KB_SOURCE_FEISHU_DOC:
            continue
        verdict = await one_shot_json(
            db,
            _ALIAS_SYSTEM,
            json.dumps(
                {
                    "question": question,
                    "article": _pair_entry(article),
                },
                ensure_ascii=False,
            )
            + '\n\n请输出 JSON:{"relation":"vocabulary|not_covered",'
            '"aliases":["员工会用的说法", ...],"reason":"中文一句话"}',
            timeout=40.0,
            max_tokens=500,
        )
        if verdict is None:
            report.llm_available = False
            return
        if str(verdict.get("relation") or "").strip().lower() != "vocabulary":
            continue
        aliases = [
            " ".join(str(item).split())
            for item in (verdict.get("aliases") or [])
            if str(item).strip()
        ][:6]
        if not aliases:
            continue
        row = await loop.propose(
            db,
            corpus=CORPUS_SUPPORT_KB,
            kind=KIND_ADD_ALIASES,
            title=f"给「{article.title}」补 {len(aliases)} 条说法",
            source=SOURCE_LINT,
            target_ref=str(article.id),
            payload={"aliases": aliases},
            rationale=str(verdict.get("reason") or "").strip()[:1000] or None,
            evidence={"employee_question": question},
        )
        if row:
            report.aliases += 1


# ---------------------------------------------------------------------------
# Pass 3 — dead links
# ---------------------------------------------------------------------------


async def lint_dead_links(db: AsyncSession, *, report: LintReport) -> None:
    """Edges pointing at a page that has since been archived or deleted.

    Produced routinely rather than exceptionally: the Feishu mirror's
    disappearance sweep archives pages every time a document is moved out of a
    watched subtree, and any link into one of those is now an edge to nothing.
    """
    links = (
        (await db.execute(select(SupportKbArticleLink))).scalars().all()
    )
    if not links:
        return
    referenced = {link.from_article_id for link in links} | {
        link.to_article_id for link in links
    }
    alive = {
        row[0]: row[1]
        for row in (
            await db.execute(
                select(SupportKbArticle.id, SupportKbArticle.status).where(
                    SupportKbArticle.id.in_(referenced)
                )
            )
        ).all()
    }
    titles = {
        row[0]: row[1]
        for row in (
            await db.execute(
                select(SupportKbArticle.id, SupportKbArticle.title).where(
                    SupportKbArticle.id.in_(referenced)
                )
            )
        ).all()
    }
    for link in links:
        for near, far in (
            (link.from_article_id, link.to_article_id),
            (link.to_article_id, link.from_article_id),
        ):
            status = alive.get(far)
            if status is not None and status != KB_ARCHIVED:
                continue
            if alive.get(near) in (None, KB_ARCHIVED):
                # Both ends gone — nothing to fix on a page nobody reads.
                continue
            row = await loop.propose(
                db,
                corpus=CORPUS_SUPPORT_KB,
                kind=KIND_DEAD_LINK,
                title=f"「{titles.get(near, near)}」指向了一条已下架/不存在的条目",
                source=SOURCE_LINT,
                target_ref=str(near),
                related_ref=str(far),
                payload={"link_id": str(link.id), "link_kind": link.kind},
                rationale=(
                    "被指向的条目已下架或已删除 —— 这条关联现在指向空处。"
                    "接受即删除这条关联。"
                ),
                evidence={"far_status": status or "missing"},
            )
            if row:
                report.dead_links += 1
            break


# ---------------------------------------------------------------------------
# Backfill — answers that worked, and evaluations that did not
# ---------------------------------------------------------------------------


async def backfill_from_answers(
    db: AsyncSession, *, window_days: int = 14, limit: int = 10,
) -> dict[str, Any]:
    """Assistant answers employees marked helpful, offered back to the corpus.

    A grounded answer the assistant synthesised from three pages, that an
    employee then said worked, is knowledge the corpus does not hold in that
    shape — and today it is thrown away and re-synthesised the next time
    somebody asks. This routes it through the same "is it already written
    down?" question as any other new knowledge, so the usual outcome is an
    alias or an appended section rather than yet another page.
    """
    from app.assistant.evaluation.service import redact  # noqa: PLC0415
    from app.assistant.models import (  # noqa: PLC0415
        ROLE_USER,
        VERDICT_HELPFUL,
        AssistantMessage,
        AssistantMessageFeedback,
    )
    from app.domains.support.kb_wiki import propose_from_knowledge  # noqa: PLC0415

    since = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=window_days)
    rated = (
        (
            await db.execute(
                select(AssistantMessageFeedback)
                .where(
                    AssistantMessageFeedback.verdict == VERDICT_HELPFUL,
                    AssistantMessageFeedback.created_at >= since,
                )
                .order_by(AssistantMessageFeedback.created_at.desc())
                .limit(max(1, min(limit, 50)))
            )
        )
        .scalars()
        .all()
    )
    considered = 0
    outcomes: dict[str, int] = {}
    for feedback in rated:
        answer = await db.get(AssistantMessage, feedback.message_id)
        if answer is None or not (answer.content or "").strip():
            continue
        asked = (
            await db.execute(
                select(AssistantMessage)
                .where(
                    AssistantMessage.conversation_id == answer.conversation_id,
                    AssistantMessage.role == ROLE_USER,
                    AssistantMessage.seq < answer.seq,
                )
                .order_by(AssistantMessage.seq.desc())
                .limit(1)
            )
        ).scalar_one_or_none()
        if asked is None or not (asked.content or "").strip():
            continue
        considered += 1
        # Scrubbed BEFORE it can become corpus, both halves.
        #
        # This answer was produced by caller-scoped read tools for ONE employee,
        # so it routinely names their device serial, their ticket number, their
        # manager, their seat list. The ``new_article`` branch already treats
        # this content as untrusted — it files a DRAFT — but ``append_section``
        # splices it into a live article that ``GET /support/kb`` serves to
        # every employee. The evaluation miner scrubs at mining time for exactly
        # this reason, and states why: a row that never held the value cannot
        # leak it later. This path is the same shape and had none.
        #
        # ⚠️ **This is not full redaction, and calling it that would be the
        # overclaim.** ``redact`` knows five patterns: e-mail, PRC mobile
        # numbers, hyphenated UUIDs, four token prefixes, and 15–19 digit runs.
        # It does NOT catch what is most likely to be in one of these answers —
        # a colleague's name, a device serial or asset tag, an employee number,
        # a ticket id, a non-PRC phone number, an office location, or a Feishu
        # open_id. What it removes is the machine-readable half; the reviewer
        # still has to READ the proposal. This is what makes a fast click
        # survivable, not what makes it safe.
        result = await propose_from_knowledge(
            db,
            title=redact(" ".join(asked.content.split()))[:200],
            body=redact(answer.content),
            category="other",
            source=SOURCE_ANSWER_BACKFILL,
            source_ref=f"message:{answer.id}",
            evidence={
                "employee_question": redact(" ".join(asked.content.split()))[:500],
                "rated": "helpful",
            },
        )
        outcomes[result.decision] = outcomes.get(result.decision, 0) + 1
    return {"considered": considered, "outcomes": outcomes}


async def backfill_from_eval_failures(
    db: AsyncSession, *, run_id: uuid.UUID, limit: int = 20,
) -> dict[str, Any]:
    """Evaluation failures the harness blamed on the corpus, as proposals.

    Only ``kb_doc`` and ``retrieval``. The other classifications are somebody
    else's work: ``tool_field_or_data`` is a data defect, ``action`` is a
    workflow, ``out_of_scope`` is a product decision, and turning any of them
    into a document would paper over the actual problem with prose.
    """
    from app.assistant.evaluation.constants import (  # noqa: PLC0415
        FAIL_KB_DOC,
        FAIL_RETRIEVAL,
        VERDICT_FAIL,
    )
    from app.assistant.evaluation.models import (  # noqa: PLC0415
        AssistantEvalResult,
    )
    from app.domains.support.kb_service import SupportKbService  # noqa: PLC0415

    rows = (
        (
            await db.execute(
                select(AssistantEvalResult)
                .where(
                    AssistantEvalResult.run_id == run_id,
                    AssistantEvalResult.verdict == VERDICT_FAIL,
                    AssistantEvalResult.failure_type.in_(
                        (FAIL_KB_DOC, FAIL_RETRIEVAL),
                    ),
                )
                .limit(max(1, min(limit, 100)))
            )
        )
        .scalars()
        .all()
    )
    service = SupportKbService(db)
    created = 0
    seen: set[str] = set()
    for row in rows:
        key = " ".join(row.question.split()).casefold()
        if key in seen:
            continue
        seen.add(key)

        if row.failure_type == FAIL_RETRIEVAL:
            # The document exists and was not found — that is a vocabulary
            # problem, and the page it should have matched is the best hit.
            found = await service.search_published(row.question, limit=1)
            if not found or found[0].source == KB_SOURCE_FEISHU_DOC:
                continue
            article = found[0]
            proposal = await loop.propose(
                db,
                corpus=CORPUS_SUPPORT_KB,
                kind=KIND_ADD_ALIASES,
                title=f"给「{article.title}」补员工的问法",
                source=SOURCE_EVAL_FAILURE,
                source_ref=f"result:{row.id}",
                target_ref=str(article.id),
                payload={"aliases": [" ".join(row.question.split())[:200]]},
                rationale=(
                    "评测判定为「检索没找到」:文档在,但用这句话问不出来。"
                    + (f"评委的说法:{row.judge_reasoning}" if row.judge_reasoning else "")
                )[:2000],
                evidence={"eval_question": row.question, "run_id": str(run_id)},
            )
        else:
            # No document covers it. Nothing here can write the document — a
            # generated one would be exactly the confident, unsupported prose
            # this whole system exists to keep out of the corpus. The proposal
            # carries the question and the reviewer writes the answer.
            proposal = await loop.propose(
                db,
                corpus=CORPUS_SUPPORT_KB,
                kind=KIND_NEW_ARTICLE,
                title=f"缺一篇文档:{row.question[:100]}",
                source=SOURCE_EVAL_FAILURE,
                source_ref=f"result:{row.id}",
                payload={
                    "title": row.question[:200],
                    "body": "",
                    "category": "other",
                    "needs_author": True,
                },
                rationale=(
                    "评测判定为「缺文档」。这条提案刻意**不带正文** —— 生成的正文"
                    "正是这套东西要挡住的那种「读起来很像、其实没依据」的内容。"
                    + (f"评委的说法:{row.judge_reasoning}" if row.judge_reasoning else "")
                )[:2000],
                evidence={"eval_question": row.question, "run_id": str(run_id)},
            )
        if proposal:
            created += 1
    return {"created": created, "failures_seen": len(rows)}


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------


async def run_lint(
    db: AsyncSession, *, window_days: int = 14, summarise_limit: int = 25,
) -> dict[str, Any]:
    """One full pass. Safe to re-run: every proposal deduplicates on content."""
    from app.domains.support.kb_wiki import summarise_missing  # noqa: PLC0415

    report = LintReport()
    summaries = await summarise_missing(db, limit=summarise_limit)
    await lint_pairs(db, report=report)
    await lint_vocabulary(db, report=report, window_days=window_days)
    await lint_dead_links(db, report=report)
    entries = await index(db, use_cache=False)
    return {
        **report.as_dict(),
        "summaries": summaries,
        "corpus_size": len(entries),
        "without_summary": sum(1 for e in entries if not e["has_summary"]),
        "without_aliases": sum(1 for e in entries if not e["aliases"]),
    }


__all__ = [
    "GAP_ATTACH_FLOOR",
    "LintReport",
    "PAIR_SIMILARITY_FLOOR",
    "backfill_from_answers",
    "backfill_from_eval_failures",
    "lint_dead_links",
    "lint_pairs",
    "lint_vocabulary",
    "run_lint",
]