"""安全 FAQ —— the employee-facing face of the 答案库. The same curated Q&A the security team maintains for customer questionnaires, projected for an audience of ~500 colleagues who are asking on their own behalf. Three things separate this projection from the admin one and none of them are cosmetic: * **Visibility.** Only ``status='active' AND employee_visible=1`` rows exist here. Nothing else can be reached, including by id. * **Field narrowing.** An employee gets the question, the answer, the category and when it was last updated. Owner, ``usage_count``, staleness, the backing controls and the source documents stay on the admin side — they are governance signals, and "待复核" printed next to a security answer starts a conversation nobody wants to have with 500 people at once. * **Language.** One language per response, resolved from the caller's UI language rather than shipping both halves and letting the frontend choose. Japanese falls back to English, matching the rest of Helix. The ask path is retrieval-augmented and deliberately grounded (``security_faq_rag``): lexical term scoring and semantic (embedding) recall run as two channels, fused by RRF, and the model is then allowed to write a short lead answer — but ONLY out of the entries that fusion returned, and only with citations the server checks against that set. **The stored entries are always rendered underneath, verbatim**, so a reader can always see what the security team actually wrote; an answer that cites nothing is discarded rather than shown. What retrieval cannot place — and what the model reads and calls unanswerable — lands in the gap ledger, which is the mechanism that makes the library grow toward what people actually ask. Naive-UTC datetime discipline (codebase-wide MySQL/TiDB convention). """ from __future__ import annotations import logging import re import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import Select, case, func, literal, or_, select, update from sqlalchemy.dialects.mysql import insert as mysql_insert from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import NotFoundError, ValidationError from app.domains.auditor import security_faq_rag as rag from app.domains.auditor.answer_freshness import STALE_CONTROL_FAILING from app.domains.auditor.inbound_ai import build_llm_client from app.domains.auditor.models import ( AuditorAnswerLibraryEntry, AuditorSecurityFaqGap, AuditorSecurityFaqVote, ) from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) # ── Language ────────────────────────────────────────────────────────── # The library stores two languages; the workforce reads three UIs. ja # resolves to en rather than to zh — the Japan office reads English # security material, and a Japanese speaker shown Chinese would have no # way to tell whether it was a fallback or the only answer we have. SUPPORTED_LANGS: tuple[str, ...] = ("zh", "en") DEFAULT_LANG = "zh" def resolve_lang(raw: str | None) -> str: """Map anything the client sends (``zh-CN``, ``ja-JP``, ``en``) to a stored language. Unknown → the default.""" token = (raw or "").strip().lower().replace("_", "-") if not token: return DEFAULT_LANG primary = token.split(",")[0].split("-")[0].strip() if primary == "zh": return "zh" if primary in ("en", "ja"): return "en" return DEFAULT_LANG # ── Category catalogue ──────────────────────────────────────────────── # ``category`` is free text on the entry (the admin form has always let # the security team coin one), so this catalogue is a *label* source, not # a constraint. Rendering happens here rather than in the frontend # because the frontend is a pure renderer and because an employee must # never be shown ``data_loss_prevention`` as if it were a word. CATEGORY_LABELS: dict[str, dict[str, str]] = { # Keys are the ones the library ACTUALLY uses (seeded by migrations 381 # and 528, verified against the real inventory) plus ``general``, the # admin form's default. Nothing speculative: a label for a key nobody # coined is a translation that can only ever be wrong-by-omission # somewhere else. "general": {"zh": "通用", "en": "General", "ja": "全般"}, "access_control": {"zh": "账号与访问控制", "en": "Access Control", "ja": "アクセス制御"}, "ai_governance": {"zh": "AI 使用规范", "en": "AI Governance", "ja": "AI ガバナンス"}, "application_security": { "zh": "应用与代码安全", "en": "Application Security", "ja": "アプリケーションセキュリティ", }, "bcp_dr": {"zh": "业务连续性与灾备", "en": "Business Continuity & DR", "ja": "事業継続と災害復旧"}, "certification": {"zh": "合规认证", "en": "Certifications", "ja": "認証"}, "data_loss_prevention": {"zh": "数据防泄漏", "en": "Data Loss Prevention", "ja": "情報漏洩対策"}, "data_privacy": {"zh": "数据隐私", "en": "Data Privacy", "ja": "データプライバシー"}, "email_security": {"zh": "邮件安全", "en": "Email Security", "ja": "メールセキュリティ"}, "encryption": {"zh": "加密与密钥", "en": "Encryption & Keys", "ja": "暗号化と鍵管理"}, "endpoint_security": {"zh": "终端安全", "en": "Endpoint Security", "ja": "エンドポイント"}, "governance": {"zh": "安全治理", "en": "Security Governance", "ja": "セキュリティガバナンス"}, "hr_security": {"zh": "人员安全", "en": "Personnel Security", "ja": "人的セキュリティ"}, "identity_lifecycle": { "zh": "账号生命周期", "en": "Identity Lifecycle", "ja": "アカウントライフサイクル", }, "incident_response": {"zh": "安全事件响应", "en": "Incident Response", "ja": "インシデント対応"}, "logging_monitoring": {"zh": "日志与监控", "en": "Logging & Monitoring", "ja": "ログと監視"}, "network_security": {"zh": "网络安全", "en": "Network Security", "ja": "ネットワーク"}, "physical_security": {"zh": "物理与办公安全", "en": "Physical Security", "ja": "物理セキュリティ"}, "risk_management": {"zh": "风险管理", "en": "Risk Management", "ja": "リスク管理"}, "supply_chain": {"zh": "供应商与第三方", "en": "Supply Chain", "ja": "サプライチェーン"}, "vulnerability_management": { "zh": "漏洞管理", "en": "Vulnerability Management", "ja": "脆弱性管理", }, } def category_label(key: str, lang: str) -> str: """Localised label for a category key. An uncatalogued key (the security team coined a new one in the admin form) is humanised — ``foo_bar`` → ``Foo Bar`` — in every language. Showing a readable English phrase beats showing a snake_case token, and beats inventing a translation we do not have. """ entry = CATEGORY_LABELS.get(key) if entry is not None: return entry.get(lang) or entry["en"] return " ".join(part.capitalize() for part in key.replace("_", " ").split()) # ── Projection ──────────────────────────────────────────────────────── _MAX_QUERY_LEN = 200 _MIN_QUERY_LEN = 2 # What goes into the model prompt when keyword search comes up empty. # Questions only — the model picks ids, it never writes the answer, so # shipping the answer bodies would buy nothing and cost the token budget. _AI_CANDIDATE_CAP = 200 _AI_MAX_MATCHES = 3 #: How deep the filter box's semantic rescue goes. It is a rescue, not a #: ranking: showing 60 loosely-related answers to somebody who typed a word #: is not better than showing none. _SEMANTIC_FILTER_DEPTH = 24 def _naive_now() -> datetime: return datetime.now(UTC).replace(tzinfo=None) def _to_naive(dt: datetime | None) -> datetime | None: if dt is None: return None if dt.tzinfo is not None: return dt.astimezone(UTC).replace(tzinfo=None) return dt def faq_view(e: AuditorAnswerLibraryEntry, *, lang: str) -> dict[str, Any]: """The narrowed, single-language shape an employee is allowed to see. Every field an employee does NOT get is a deliberate omission: owner, usage_count, is_stale / stale_reason, review cadence, source_doc_ids and backing_control_keys are all admin-side governance state. """ zh = lang == "zh" question = (e.question_zh if zh else e.question_en) or e.question_en or e.question_zh answer = (e.answer_zh if zh else e.answer_en) or e.answer_en or e.answer_zh updated = _to_naive(e.updated_at) return { "id": str(e.id), "category": e.category, "category_label": category_label(e.category, lang), "question": question, "answer": answer, "tags": e.tags or [], "is_featured": bool(e.is_featured), # 有帮助 only. The 没帮助 tally stays on the admin side with the # rest of the governance state: a running public count of people # who marked a COMPANY SECURITY RULE unhelpful reads as a verdict # on the rule rather than on the writing, and it is visible to the # next 500 readers before they have formed their own. The button # still works and still shows the reader their own vote — the # scoreboard is for the team that has to act on it. "helpful_count": e.employee_helpful_count, "updated_at": updated.isoformat() if updated else None, } def _visible_base() -> Select: return select(AuditorAnswerLibraryEntry).where( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(True), ) # LIKE metacharacters are escaped with ``/`` rather than the SQL default # backslash: a backslash escape has to survive both the Python literal and # MySQL's own string parsing (and changes meaning under # ``NO_BACKSLASH_ESCAPES``), which is three chances to get one character # wrong. ``/`` never appears in a LIKE pattern by accident. _LIKE_ESCAPE = "/" def _like(term: str) -> str: """A containment pattern that means the characters the reader typed. Unescaped, ``%`` and ``_`` are wildcards: searching ``%`` returned the ENTIRE library and ``a%b`` matched 63 of 71 entries that contain no such string — measured, not theorised. Answers here quote SLAs and percentages ("99.9%") and category keys ("data_loss_prevention"), so both characters are ones a reader really types. """ escaped = ( term.replace(_LIKE_ESCAPE, _LIKE_ESCAPE * 2) .replace("%", f"{_LIKE_ESCAPE}%") .replace("_", f"{_LIKE_ESCAPE}_") ) return f"%{escaped}%" def _search_clause(search: str): """Match across BOTH stored languages regardless of UI language. Somebody on the Chinese UI searching ``USB`` or ``MFA`` must find the entry even when only its English half spells it that way, and the same in reverse. Narrowing the search to the display language would silently hide answers we have. """ like = _like(search) return or_( AuditorAnswerLibraryEntry.question_zh.ilike(like, escape=_LIKE_ESCAPE), AuditorAnswerLibraryEntry.question_en.ilike(like, escape=_LIKE_ESCAPE), # The employee's own words for this entry (migration 594). Searched # because that is the entire point of the column: the library is # written in questionnaire register and the reader is not. AuditorAnswerLibraryEntry.employee_phrasings.ilike( like, escape=_LIKE_ESCAPE, ), AuditorAnswerLibraryEntry.answer_zh.ilike(like, escape=_LIKE_ESCAPE), AuditorAnswerLibraryEntry.answer_en.ilike(like, escape=_LIKE_ESCAPE), ) def _relevance_order(search: str): """A question hit outranks an answer hit. Without this, searching 「离职」 puts every answer that merely mentions offboarding above the entry actually titled with it. """ like = _like(search) return case( (AuditorAnswerLibraryEntry.question_zh.ilike(like, escape=_LIKE_ESCAPE), 0), (AuditorAnswerLibraryEntry.question_en.ilike(like, escape=_LIKE_ESCAPE), 0), # An employee-phrasing hit ranks WITH the question hits, not with the # answer hits: somebody wrote that phrasing down precisely because it # is how this entry gets asked for. ( AuditorAnswerLibraryEntry.employee_phrasings.ilike( like, escape=_LIKE_ESCAPE, ), 0, ), else_=1, ) def normalize_query(raw: str) -> str: """Dedup key for the gap ledger: collapsed whitespace, lower-cased, trimmed to the column width.""" return re.sub(r"\s+", " ", (raw or "").strip()).lower()[:255] # ── Reads ───────────────────────────────────────────────────────────── async def list_faq( db: AsyncSession, *, lang: str, category: str | None = None, search: str | None = None, page: int = 1, page_size: int = 20, ) -> dict[str, Any]: base = _visible_base() term = (search or "").strip() if category: base = base.where(AuditorAnswerLibraryEntry.category == category) async def _count(query) -> int: return int((await db.execute( select(func.count()).select_from(query.subquery()) )).scalar_one()) ordering: tuple[Any, ...] if not term: stmt = base total = await _count(stmt) ordering = ( AuditorAnswerLibraryEntry.is_featured.desc(), AuditorAnswerLibraryEntry.employee_view_count.desc(), AuditorAnswerLibraryEntry.updated_at.desc(), ) else: # Substring first: when the reader typed an actual keyword it is # exact and predictable, which is what a filter box should be. stmt = base.where(_search_clause(term)) total = await _count(stmt) ordering = ( _relevance_order(term), AuditorAnswerLibraryEntry.employee_view_count.desc(), AuditorAnswerLibraryEntry.updated_at.desc(), ) scored = _term_match(term) if total == 0 else None if scored is not None: # Nothing matched as a substring. Rather than show 「没有匹配」 # for a question we can answer — 「钓鱼邮件」 appears nowhere in # the entry about 防钓鱼 — fall back to the same term scoring # the ask box uses. An empty list is the one result a reader # cannot act on. matched, weighted, required = scored fallback = base.where(matched >= required) fallback_total = await _count(fallback) if fallback_total: stmt, total = fallback, fallback_total ordering = ( weighted.desc(), matched.desc(), AuditorAnswerLibraryEntry.employee_view_count.desc(), ) if total == 0: # Still nothing: not one shared character run between what the # reader typed and a library written in questionnaire voice. # This is exactly where a filter box has always given up and # where semantic recall earns its keep — 「个人电脑能不能写代码」 # shares no term with 「员工能否使用个人设备访问公司数据?」. # # Ranked in Python, not SQL: the candidate set is bounded by # ``_SEMANTIC_FILTER_DEPTH``, so ordering by cosine costs a # slice, while expressing that order as a CASE over UUID binds # would be a driver-level bind question for no gain. semantic = await rag.semantic_ranked( db, query=term, limit=_SEMANTIC_FILTER_DEPTH, ) ids = [entry_id for entry_id, _ in (semantic or [])] if ids: found = (await db.execute( base.where(AuditorAnswerLibraryEntry.id.in_(ids)) )).scalars().all() by_id = {e.id: e for e in found} ranked = [by_id[i] for i in ids if i in by_id] start = (page - 1) * page_size return { "items": [ faq_view(r, lang=lang) for r in ranked[start:start + page_size] ], "meta": { "page": page, "page_size": page_size, "total": len(ranked), }, } rows = (await db.execute( stmt.order_by(*ordering) .offset((page - 1) * page_size).limit(page_size) )).scalars().all() return { "items": [faq_view(r, lang=lang) for r in rows], "meta": {"page": page, "page_size": page_size, "total": total}, } async def facets(db: AsyncSession, *, lang: str) -> dict[str, Any]: """Category counts + the featured set, computed over the WHOLE visible library rather than over whatever the caller is currently filtering. Deriving the filter options from the filtered result is how a filter UI ends up able to narrow but never widen. """ counts = (await db.execute( select(AuditorAnswerLibraryEntry.category, func.count()).where( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(True), ).group_by(AuditorAnswerLibraryEntry.category) )).all() categories = sorted( ( { "key": key, "label": category_label(key, lang), "count": int(count), } for key, count in counts ), key=lambda c: (-c["count"], c["label"]), ) featured_rows = (await db.execute( _visible_base().where( AuditorAnswerLibraryEntry.is_featured.is_(True), ).order_by( AuditorAnswerLibraryEntry.employee_view_count.desc(), AuditorAnswerLibraryEntry.updated_at.desc(), ).limit(10) )).scalars().all() last_updated = (await db.execute( select(func.max(AuditorAnswerLibraryEntry.updated_at)).where( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(True), ) )).scalar_one_or_none() last_updated = _to_naive(last_updated) return { "total": sum(c["count"] for c in categories), "categories": categories, "featured": [faq_view(r, lang=lang) for r in featured_rows], "last_updated_at": last_updated.isoformat() if last_updated else None, } async def get_visible_entry( db: AsyncSession, entry_id: uuid.UUID, ) -> AuditorAnswerLibraryEntry: """One entry by id, or 404 — including when it exists but is hidden. A retracted entry must be indistinguishable from one that never existed, otherwise the deep link an employee already has keeps working as an existence oracle after the security team pulls it. """ e: AuditorAnswerLibraryEntry | None = (await db.execute( _visible_base().where(AuditorAnswerLibraryEntry.id == entry_id) )).scalar_one_or_none() if e is None: raise NotFoundError("security FAQ entry not found") return e # ── Engagement ──────────────────────────────────────────────────────── async def record_view(db: AsyncSession, entry_id: uuid.UUID) -> None: """Bump the employee read counter with a single atomic UPDATE. Read-modify-write would lose counts whenever two people open the same answer at once, which for the top entries is the normal case. """ await db.execute( update(AuditorAnswerLibraryEntry) .where( AuditorAnswerLibraryEntry.id == entry_id, AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(True), ) .values( employee_view_count=( AuditorAnswerLibraryEntry.employee_view_count + 1 ), ) .execution_options(synchronize_session=False) ) _VIEW_DEDUP_TTL_SECONDS = 24 * 3600 async def record_view_once( db: AsyncSession, *, entry_id: uuid.UUID, viewer_id: uuid.UUID, ) -> bool: """Count a read at most once per person per entry per day. Without the de-duplication the counter measures accordion clicks, and the "top questions" list it feeds would rank whoever fidgets most. A Redis outage falls open (the read is counted): an over-count in a soft popularity metric is a better failure than dropping the signal. """ key = f"security_faq_view:{viewer_id}:{entry_id}" try: from app.core.redis_client import get_redis first = await get_redis().set( key, "1", ex=_VIEW_DEDUP_TTL_SECONDS, nx=True, ) if not first: return False except Exception as exc: # noqa: BLE001 — soft metric, never block a read logger.warning("security FAQ view dedup unavailable (%s)", exc) await record_view(db, entry_id) return True async def record_vote( db: AsyncSession, *, entry_id: uuid.UUID, person_id: uuid.UUID, helpful: bool, ) -> dict[str, Any]: """Cast or change one person's verdict on one entry. The vote row is the source of truth; the counters on the entry are a denormalisation kept correct by moving both in one transaction. A person who flips their vote moves the count from one column to the other rather than adding to both. Casting the same verdict twice is a no-op, which is what makes a double-tap safe without a client-side guard: the second request loses the unique index, re-reads the winner's row and returns. The returned count is this transaction's view. Two DIFFERENT people voting in the same instant each see their own increment applied to the snapshot they started from, so a reader can briefly see a total one short of the truth; the stored value is exact either way because the counters move with ``col = col + 1`` in SQL, not read-modify-write. """ entry = await get_visible_entry(db, entry_id) existing = (await db.execute( select(AuditorSecurityFaqVote).where( AuditorSecurityFaqVote.entry_id == entry_id, AuditorSecurityFaqVote.person_id == person_id, ) )).scalar_one_or_none() if existing is None: try: # add() INSIDE the savepoint: begin_nested() autoflushes # first, so a row added before it would collide outside the # savepoint and poison the whole transaction. async with db.begin_nested(): db.add(AuditorSecurityFaqVote( entry_id=entry_id, person_id=person_id, helpful=helpful, )) await db.flush() except IntegrityError: # Somebody else's request for the same (entry, person) won the # race. Re-read with a locking read: this transaction's # snapshot predates their commit, so a plain read would still # return None. existing = (await db.execute( select(AuditorSecurityFaqVote).where( AuditorSecurityFaqVote.entry_id == entry_id, AuditorSecurityFaqVote.person_id == person_id, ).with_for_update().execution_options(populate_existing=True) )).scalar_one() # The winner also moved the counters, and ``entry`` was read # before they committed. Under REPEATABLE READ a plain refresh # would return this transaction's stale snapshot, so the caller # would be handed — and would repaint — a count one short. await db.refresh(entry, with_for_update=True) else: await _shift_vote_counters(db, entry_id, add_helpful=helpful) await db.flush() await db.refresh(entry) return _vote_view(entry, helpful) if existing.helpful == helpful: return _vote_view(entry, helpful) existing.helpful = helpful await _shift_vote_counters( db, entry_id, add_helpful=helpful, remove_helpful=not helpful, ) await db.flush() await db.refresh(entry) return _vote_view(entry, helpful) def _vote_view(e: AuditorAnswerLibraryEntry, helpful: bool) -> dict[str, Any]: """Same narrowing as ``faq_view``: the caller's own verdict, and the positive tally only.""" return { "id": str(e.id), "helpful": helpful, "helpful_count": e.employee_helpful_count, } async def _shift_vote_counters( db: AsyncSession, entry_id: uuid.UUID, *, add_helpful: bool, remove_helpful: bool | None = None, ) -> None: """Move the denormalised counters, never below zero. The GREATEST() floor is not defensive noise: counters that predate a vote row (or survive one being deleted by a future data fix) would otherwise be able to go negative and render as ``-1 人觉得有帮助``. """ add_col = ( AuditorAnswerLibraryEntry.employee_helpful_count if add_helpful else AuditorAnswerLibraryEntry.employee_unhelpful_count ) values: dict[str, Any] = {add_col.key: add_col + 1} if remove_helpful is not None: rm_col = ( AuditorAnswerLibraryEntry.employee_helpful_count if remove_helpful else AuditorAnswerLibraryEntry.employee_unhelpful_count ) values[rm_col.key] = func.greatest(rm_col - 1, 0) await db.execute( update(AuditorAnswerLibraryEntry) .where(AuditorAnswerLibraryEntry.id == entry_id) .values(**values) .execution_options(synchronize_session=False) ) async def my_votes( db: AsyncSession, *, person_id: uuid.UUID, entry_ids: list[uuid.UUID], ) -> dict[str, bool]: """``{entry_id: helpful}`` for the entries the caller has voted on, so the page can show their own verdict instead of an unlit button.""" if not entry_ids: return {} rows = (await db.execute( select( AuditorSecurityFaqVote.entry_id, AuditorSecurityFaqVote.helpful, ).where( AuditorSecurityFaqVote.person_id == person_id, AuditorSecurityFaqVote.entry_id.in_(entry_ids), ) )).all() return {str(entry_id): bool(helpful) for entry_id, helpful in rows} # ── Ask ─────────────────────────────────────────────────────────────── #: How many candidates each retrieval channel contributes to the fusion. #: Deeper than the number shown: RRF's whole value is that a document both #: channels rank *reasonably* outranks one that tops a single channel, and a #: depth of 5 would never let the two channels disagree usefully. _CHANNEL_DEPTH = 20 async def ask( db: AsyncSession, *, query: str, lang: str, person_id: uuid.UUID | None, limit: int = 5, compose: bool = True, record_gaps: bool = True, ) -> dict[str, Any]: """Answer a free-text question strictly out of the visible library. Three stages, each of which degrades into the next: 1. **Retrieval.** Lexical term scoring and semantic (embedding) recall run as independent channels and are fused by RRF. Either channel alone still answers — no LLM key, or nothing embedded yet, and the page behaves exactly as it did before this existed. 2. **Composition.** The retrieved entries go back to the model, which writes a short lead answer *out of them* and cites the ones it used. Ungrounded prose is dropped server-side; the stored entries are rendered regardless. 3. **The gap ledger.** A question that retrieval could not place — or that the model read the retrieved entries and called unanswerable — is recorded. That second case is the one the old shape could never see: retrieval almost always returns *something*, so "we showed them five answers and none of them was theirs" used to look identical to a hit. """ term = (query or "").strip() if len(term) < _MIN_QUERY_LEN: raise ValidationError("question is too short") term = term[:_MAX_QUERY_LEN] lexical = await _keyword_hits(db, term=term, limit=_CHANNEL_DEPTH) semantic = await rag.semantic_ranked(db, query=term, limit=_CHANNEL_DEPTH) channels: dict[str, list[uuid.UUID]] = {} if lexical: channels["keyword"] = [e.id for e in lexical] if semantic: channels["semantic"] = [entry_id for entry_id, _ in semantic] if channels: fused = rag.rrf_fuse(channels, limit=limit) # Re-read through the visibility gate rather than trusting the # vector cache's id set: an entry retracted since the cache was # built must not be renderable, and this is the only place that # can guarantee it. entries = await _entries_in_order(db, [entry_id for entry_id, _, _ in fused]) if entries: matched_by = ( "hybrid" if len(channels) > 1 else next(iter(channels)) ) found_in = {entry_id: chans for entry_id, _, chans in fused} # How sure retrieval is, per entry, from what it already knows: # the cosine it ranked by and whether the two channels agreed. cosines = {entry_id: score for entry_id, score in (semantic or [])} lex_rank = {e.id: i for i, e in enumerate(lexical or [], start=1)} tiers = { e.id: rag.confidence_tier( channels=found_in.get(e.id, []), cosine=cosines.get(e.id), lexical_rank=lex_rank.get(e.id), ) for e in entries } best_tier = _best_tier(tiers.values()) # The composition floor. A model asked to summarise material the # retrieval itself rates weak is exactly where it starts # stretching — and a stretched sentence here is read as company # policy. Below the floor the reader still gets the entries, with # their confidence shown, and can judge for themselves. answer = ( await _compose(db, term=term, entries=entries, lang=lang) if (compose and best_tier != rag.TIER_LOW) else None ) # Rerank: drop what the model read and called off-topic. Never # below one row — an empty list would turn a hit into a dead end # on the strength of one model call. relevant = (answer or {}).get("relevant_ids") if relevant: kept = [e for e in entries if str(e.id) in set(relevant)] if kept: entries = kept covered = answer is None or answer.get("covered") is not False asked_before = None if not covered and record_gaps: # Retrieval found neighbours; the model says none of them # answers this. That is a gap, and a more informative one # than a zero-hit query. asked_before = await record_gap( db, query=term, lang=lang, person_id=person_id, ) return { "matched_by": matched_by, "query": term, "items": [faq_view(e, lang=lang) for e in entries], "answer": ( answer if (answer and answer.get("covered")) else None ), "covered": covered, "ask_count": asked_before, "retrieval": { "keyword": len(channels.get("keyword", [])), "semantic": len(channels.get("semantic", [])), "semantic_available": semantic is not None, "channels": { str(e.id): found_in.get(e.id, []) for e in entries }, "confidence": { str(e.id): tiers.get(e.id, rag.TIER_LOW) for e in entries }, "best_confidence": best_tier, "reranked": bool(relevant), }, } # Nothing retrieved at all. The id-picker is what answers when there # are no vectors yet (a fresh deployment, a key configured after the # library was seeded) — it needs no index, only the model. ai_hits, ai_available = await _ai_hits(db, term=term, lang=lang) if ai_hits: return { "matched_by": "ai", "query": term, "items": [faq_view(e, lang=lang) for e in ai_hits], "answer": None, "covered": True, "ask_count": None, "retrieval": { "keyword": 0, "semantic": 0, "semantic_available": semantic is not None, "channels": {}, "confidence": {}, "best_confidence": None, "reranked": False, }, } asked_before = ( await record_gap(db, query=term, lang=lang, person_id=person_id) if record_gaps else None ) return { "matched_by": "none", "query": term, "items": [], "answer": None, "covered": False, "ask_count": asked_before, "ai_available": ai_available, "retrieval": { "keyword": 0, "semantic": 0, "semantic_available": semantic is not None, "channels": {}, "confidence": {}, "best_confidence": None, "reranked": False, }, } _TIER_ORDER = (rag.TIER_HIGH, rag.TIER_MEDIUM, rag.TIER_LOW) def _best_tier(tiers) -> str: """The strongest tier in a result set — what the composition floor reads. Best, not average: one entry the retrieval is sure about is enough to ground a lead answer, and averaging would let a long tail of weak neighbours veto it. """ present = set(tiers) for tier in _TIER_ORDER: if tier in present: return tier return rag.TIER_LOW async def _entries_in_order( db: AsyncSession, ids: list[uuid.UUID], ) -> list[AuditorAnswerLibraryEntry]: """Load the fused ids as visible entries, in the fused order.""" if not ids: return [] rows = (await db.execute( _visible_base().where(AuditorAnswerLibraryEntry.id.in_(ids)) )).scalars().all() by_id = {e.id: e for e in rows} return [by_id[i] for i in ids if i in by_id] async def _compose( db: AsyncSession, *, term: str, entries: list[AuditorAnswerLibraryEntry], lang: str, ) -> dict[str, Any] | None: """The grounded lead answer, or None when the model is unavailable. The client is built HERE rather than inside the composer so this module keeps one seam onto the provider — the same ``build_llm_client`` the id-picker uses, so a deployment with no key never reaches either. """ try: client = await build_llm_client(db, timeout=rag.COMPOSE_TIMEOUT) except Exception as exc: # noqa: BLE001 — degradation, not failure logger.info("security FAQ ask: composition unavailable (%s)", exc) return None return await rag.compose_answer( db, query=term, entries=entries, lang=lang, client=client, ) # Latin words worth matching on; a one-character token matches everything. _LATIN_TOKEN = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.\-]*") # CJK runs (Han + kana). No spaces, so these are shingled rather than split. _CJK_RUN = re.compile(r"[\u4e00-\u9fff\u3040-\u30ff]+") # Bound the SQL: each term costs four LIKEs per row. _MAX_TERMS = 8 def query_terms(query: str) -> list[str]: """Split a natural-language question into terms to match on. A whole question is never a substring of an answer, which is the difference between a search box and an ask box. 「钓鱼邮件」 does not appear anywhere in the entry that answers it — the entry says 「防钓鱼」 and 「钓鱼识别」 — so substring matching returns nothing and the question goes to the model, or to the gap ledger, for an answer we already have. Measured against the real seeded library: three of four naturally-phrased questions found nothing. Latin runs split on word boundaries. Chinese and Japanese have none, so a CJK run becomes overlapping two-character shingles: 「钓鱼邮件」 → 钓鱼 / 鱼邮 / 邮件, of which the real entry matches two. Two characters is the shortest unit that still carries meaning in Chinese; single characters match half the library. """ terms: list[str] = [] for word in _LATIN_TOKEN.findall(query): if len(word) >= 2: terms.append(word.lower()) for run in _CJK_RUN.findall(query): if len(run) < 2: continue if len(run) == 2: terms.append(run) else: terms.extend(run[i:i + 2] for i in range(len(run) - 1)) seen: set[str] = set() out: list[str] = [] for term in terms: if term not in seen: seen.add(term) out.append(term) return out[:_MAX_TERMS] def _term_hit(term: str, *, question_only: bool): # ``_LATIN_TOKEN`` admits ``_`` (so ``data_loss_prevention`` stays one # term), which is also the single-character LIKE wildcard — without # escaping, that term would match ``dataXlossXprevention`` too. like = _like(term) cols = ( ( AuditorAnswerLibraryEntry.question_zh, AuditorAnswerLibraryEntry.question_en, # ``employee_phrasings`` counts as a QUESTION column, and so gets # the same triple weight (migration 594). Somebody wrote that # phrasing down precisely because it is how this entry gets asked # for — it carries more signal than the entry's own questionnaire # wording, not less. # # This is the path that matters: ``ask`` (the employee page and the # assistant's security tool) scores terms through here, while # ``list_faq`` searches its own clause. Wiring the column into only # one of them would leave it helping the browse list and doing # nothing for the thing that actually answers questions. AuditorAnswerLibraryEntry.employee_phrasings, ) if question_only else (AuditorAnswerLibraryEntry.answer_zh, AuditorAnswerLibraryEntry.answer_en) ) return case( (or_(*(c.ilike(like, escape=_LIKE_ESCAPE) for c in cols)), 1), else_=0, ) def _term_match(query: str): """``(matched_terms, weighted_score, required)`` or ``None``. ``matched_terms`` counts distinct terms found anywhere in an entry; ``weighted_score`` is the same with question hits tripled. ``None`` when the query yields no usable term at all (a single character, or pure punctuation). """ terms = query_terms(query) if not terms: return None matched = sum( ( case( ( or_( _term_hit(t, question_only=True) == 1, _term_hit(t, question_only=False) == 1, ), 1, ), else_=0, ) for t in terms ), literal(0), ) weighted = sum( ( _term_hit(t, question_only=True) * 3 + _term_hit(t, question_only=False) for t in terms ), literal(0), ) # One term out of eight is how 「公司」 returns half the library. required = 1 if len(terms) == 1 else 2 return matched, weighted, required async def _keyword_hits( db: AsyncSession, *, term: str, limit: int, ) -> list[AuditorAnswerLibraryEntry]: """Term-scored retrieval over the visible library. Each term that appears anywhere in an entry scores it; a term in the QUESTION is worth three times the same term in the answer, because a question that shares words with yours is far more likely to be yours than a long answer that mentions them in passing. The bar is two distinct terms (one, when the query only produced one — 「MFA」 is a complete question). One term out of eight is how 「公司」 would return half the library. Full scan by design: the library is a few hundred rows of curated prose, and a LIKE per term per column over that is far cheaper than any index that would have to be maintained for it. """ scored = _term_match(term) if scored is None: return [] matched, weighted, required = scored return list((await db.execute( _visible_base() .where(matched >= required) .order_by( weighted.desc(), matched.desc(), AuditorAnswerLibraryEntry.employee_view_count.desc(), ) .limit(limit) )).scalars().all()) _AI_SYSTEM = ( "你是公司内部的信息安全知识库检索助手。给定一名员工的提问和一份" "『候选条目』列表(每条含 id 与标准问题),请从候选条目中挑出真正能回答" "该提问的条目 id,最多 3 条,按相关度从高到低排列。" "严格要求:① 只能返回候选列表中出现过的 id;② 不要编写任何答案文本;" "③ 如果没有任何条目能回答这个提问,返回空数组 —— 宁可返回空,也不要" "返回勉强相关的条目。严格输出单个 JSON 对象。" ) async def _ai_hits( db: AsyncSession, *, term: str, lang: str, ) -> tuple[list[AuditorAnswerLibraryEntry], bool]: """``(entries, ai_available)``. Every failure mode — no LLM key configured, provider outage, unparseable reply — degrades to "no AI match" rather than an error. The employee asked a question; a 502 is not an answer, and the page still has a working search box and a route to IT. """ candidates = list((await db.execute( _visible_base().order_by( AuditorAnswerLibraryEntry.employee_view_count.desc(), AuditorAnswerLibraryEntry.usage_count.desc(), ).limit(_AI_CANDIDATE_CAP) )).scalars().all()) if not candidates: return [], False try: client = await build_llm_client(db) except Exception as exc: # noqa: BLE001 — degradation, not failure logger.info("security FAQ ask: LLM unavailable (%s)", exc) return [], False by_id = {str(e.id): e for e in candidates} payload = [ { "id": str(e.id), "question": ( e.question_zh if lang == "zh" else e.question_en ) or e.question_en, "category": e.category, } for e in candidates ] user_prompt = ( "候选条目(JSON 数组):\n" + _dump(payload) + f"\n\n员工提问:{term}\n\n" + '请输出 JSON:{"ids":["候选条目id", ...]}' ) try: resp = await client.chat_json( system=_AI_SYSTEM, user=user_prompt, max_tokens=300, ) except Exception as exc: # noqa: BLE001 — degradation, not failure logger.warning("security FAQ ask: LLM call failed (%s)", exc) return [], True ids = (resp or {}).get("ids") or [] if not isinstance(ids, list): return [], True matched: list[AuditorAnswerLibraryEntry] = [] seen: set[str] = set() for raw in ids[:_AI_MAX_MATCHES]: key = str(raw).strip() entry = by_id.get(key) if entry is not None and key not in seen: seen.add(key) matched.append(entry) return matched, True def _dump(obj: Any) -> str: import json return json.dumps(obj, ensure_ascii=False) # ── Gap ledger ──────────────────────────────────────────────────────── async def record_gap( db: AsyncSession, *, query: str, lang: str, person_id: uuid.UUID | None, ) -> int | None: """Upsert the unanswered question by normalised text, returning how many times it has now been asked (``None`` when the query was too short to record). The count goes back to the person who just asked, because a dead end that says 「你是第 7 个问这句话的人,它排在队列前面」 is a different experience from one that says 「没有找到」 —— the first tells them the system heard them and how their question ranks; the second reads as a shrug. It is an aggregate over one normalised question, so it says nothing about who else asked. The update clause bumps counters and the "who asked last" fields and touches nothing in the resolution family (status / resolved_entry_id / resolved_by_person_id / resolved_at / note). An admin's decision survives every later ask — a dismissed gap stays dismissed, and an answered one that keeps being asked shows up as exactly that rather than silently reopening. """ norm = normalize_query(query) if len(norm) < _MIN_QUERY_LEN: return None now = _naive_now() stmt = mysql_insert(AuditorSecurityFaqGap).values( id=uuid.uuid4(), query_norm=norm, query_sample=query.strip()[:_MAX_QUERY_LEN], lang=lang, ask_count=1, first_asked_at=now, last_asked_at=now, last_asker_person_id=person_id, status="open", ) await db.execute(stmt.on_duplicate_key_update( ask_count=AuditorSecurityFaqGap.ask_count + 1, last_asked_at=stmt.inserted.last_asked_at, last_asker_person_id=stmt.inserted.last_asker_person_id, query_sample=stmt.inserted.query_sample, lang=stmt.inserted.lang, updated_at=now, )) # Re-read rather than trusting the insert: on a duplicate the statement # incremented a value this session never saw, and the count shown to the # asker has to be the stored one. return (await db.execute( select(AuditorSecurityFaqGap.ask_count).where( AuditorSecurityFaqGap.query_norm == norm, ) )).scalar_one_or_none() GAP_STATUSES: tuple[str, ...] = ("open", "answered", "dismissed") def gap_view( g: AuditorSecurityFaqGap, *, asker: dict[str, Any] | None = None, entry_question: str | None = None, ) -> dict[str, Any]: resolved_at = _to_naive(g.resolved_at) last_asked = _to_naive(g.last_asked_at) first_asked = _to_naive(g.first_asked_at) return { "id": str(g.id), "query": g.query_sample, "query_norm": g.query_norm, "lang": g.lang, "ask_count": g.ask_count, "first_asked_at": first_asked.isoformat() if first_asked else None, "last_asked_at": last_asked.isoformat() if last_asked else None, "last_asker": asker, "status": g.status, "resolved_entry_id": ( str(g.resolved_entry_id) if g.resolved_entry_id else None ), "resolved_entry_question": entry_question, "resolved_at": resolved_at.isoformat() if resolved_at else None, "note": g.note, # Answered/dismissed and still being asked. The single most # actionable row in the ledger: whatever we wrote did not land. "reasked_after_resolution": bool( g.status != "open" and resolved_at is not None and last_asked is not None and last_asked > resolved_at ), } async def list_gaps( db: AsyncSession, *, status: str | None = "open", page: int = 1, page_size: int = 50, ) -> dict[str, Any]: stmt = select(AuditorSecurityFaqGap) if status: if status not in GAP_STATUSES: raise ValidationError(f"status must be one of {GAP_STATUSES}") stmt = stmt.where(AuditorSecurityFaqGap.status == status) total = (await db.execute( select(func.count()).select_from(stmt.subquery()) )).scalar_one() rows = (await db.execute( stmt.order_by( AuditorSecurityFaqGap.ask_count.desc(), AuditorSecurityFaqGap.last_asked_at.desc(), ).offset((page - 1) * page_size).limit(page_size) )).scalars().all() person_ids = {g.last_asker_person_id for g in rows if g.last_asker_person_id} people: dict[uuid.UUID, IdentityPerson] = {} if person_ids: people = { p.id: p for p in (await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(list(person_ids))) )).scalars().all() } entry_ids = {g.resolved_entry_id for g in rows if g.resolved_entry_id} entries: dict[uuid.UUID, AuditorAnswerLibraryEntry] = {} if entry_ids: entries = { e.id: e for e in (await db.execute( select(AuditorAnswerLibraryEntry).where( AuditorAnswerLibraryEntry.id.in_(list(entry_ids)) ) )).scalars().all() } def _asker(g: AuditorSecurityFaqGap) -> dict[str, Any] | None: p = people.get(g.last_asker_person_id) if g.last_asker_person_id else None if p is None: return None return { "id": str(p.id), "display_name": p.display_name, "work_email": p.work_email, } def _question(g: AuditorSecurityFaqGap) -> str | None: e = entries.get(g.resolved_entry_id) if g.resolved_entry_id else None return e.question_zh if e is not None else None return { "items": [ gap_view(g, asker=_asker(g), entry_question=_question(g)) for g in rows ], "meta": {"page": page, "page_size": page_size, "total": total}, } async def get_gap(db: AsyncSession, gap_id: uuid.UUID) -> AuditorSecurityFaqGap: g = await db.get(AuditorSecurityFaqGap, gap_id) if g is None: raise NotFoundError("security FAQ gap not found") return g async def resolve_gap( db: AsyncSession, gap_id: uuid.UUID, *, status: str, entry_id: uuid.UUID | None, note: str | None, actor_id: uuid.UUID | None, ) -> AuditorSecurityFaqGap: """Close a gap as answered (optionally linking the entry that answers it) or dismissed, or reopen it.""" if status not in GAP_STATUSES: raise ValidationError(f"status must be one of {GAP_STATUSES}") g = await get_gap(db, gap_id) if entry_id is not None: # Link only to an entry an employee can actually reach — linking a # hidden or archived entry would mark the gap answered by # something nobody can read. await get_visible_entry(db, entry_id) g.status = status g.note = (note or None) if status == "open": g.resolved_entry_id = None g.resolved_by_person_id = None g.resolved_at = None else: g.resolved_entry_id = entry_id g.resolved_by_person_id = actor_id g.resolved_at = _naive_now() await db.flush() return g async def gap_overview(db: AsyncSession) -> dict[str, Any]: counts: dict[str, int] = { str(status): int(count) for status, count in (await db.execute( select(AuditorSecurityFaqGap.status, func.count()).group_by( AuditorSecurityFaqGap.status, ) )).all() } return { "open": int(counts.get("open", 0)), "answered": int(counts.get("answered", 0)), "dismissed": int(counts.get("dismissed", 0)), } # ── Admin-side employee stats ───────────────────────────────────────── async def employee_overview(db: AsyncSession) -> dict[str, Any]: """What the employee face currently looks like, for the console. ``hidden`` counts active entries the security team has retracted — the number that answers "why does the employee page show fewer than the library holds". """ visible = (await db.execute( select(func.count()).select_from(AuditorAnswerLibraryEntry).where( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(True), ) )).scalar_one() hidden = (await db.execute( select(func.count()).select_from(AuditorAnswerLibraryEntry).where( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(False), ) )).scalar_one() featured = (await db.execute( select(func.count()).select_from(AuditorAnswerLibraryEntry).where( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(True), AuditorAnswerLibraryEntry.is_featured.is_(True), ) )).scalar_one() views = (await db.execute( select(func.coalesce( func.sum(AuditorAnswerLibraryEntry.employee_view_count), 0, )).select_from(AuditorAnswerLibraryEntry).where( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(True), ) )).scalar_one() top_rows = (await db.execute( _visible_base().where( AuditorAnswerLibraryEntry.employee_view_count > 0, ).order_by( AuditorAnswerLibraryEntry.employee_view_count.desc(), ).limit(10) )).scalars().all() # Answers people read and then told us did not help. Ranked by the # gap between the two counts, not by raw unhelpful votes, so a widely # read entry with a few dissenters does not crowd out a short one # that nobody found useful. unhelpful_rows = (await db.execute( _visible_base().where( AuditorAnswerLibraryEntry.employee_unhelpful_count > AuditorAnswerLibraryEntry.employee_helpful_count, AuditorAnswerLibraryEntry.employee_unhelpful_count > 0, ).order_by( ( AuditorAnswerLibraryEntry.employee_unhelpful_count - AuditorAnswerLibraryEntry.employee_helpful_count ).desc(), ).limit(10) )).scalars().all() # Published to 500 people, and the control the claim rests on is # currently failing. This is the one bucket where the employee face # and the platform's own posture DISAGREE — the answer says "we # enforce X" while ``auditor.answer_library_freshness`` has just # recorded that X stopped holding. # # NOT auto-hidden. A flapping control would silently yank answers off # the employee page and leave the reader with nothing, which is worse # than a stale answer; and most entries are policy statements that a # red control does not falsify. Retraction is a judgement, so this # surfaces the rows that need one instead of making it. at_risk_rows = (await db.execute( _visible_base().where( AuditorAnswerLibraryEntry.stale_reason == STALE_CONTROL_FAILING, ).order_by( AuditorAnswerLibraryEntry.employee_view_count.desc(), ).limit(10) )).scalars().all() # The 待放出 queue: active, hidden, and never yet published. Almost # all of these arrive from questionnaire backfill, which lands hidden # on purpose (single-language customer prose). Counting them here is # what stops "hidden by default" from meaning "lost". pending_rows = (await db.execute( select(AuditorAnswerLibraryEntry).where( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(False), ).order_by( AuditorAnswerLibraryEntry.created_at.desc(), ).limit(10) )).scalars().all() def _brief(e: AuditorAnswerLibraryEntry) -> dict[str, Any]: # Both languages: the console is localised like the rest of the # admin surface, and a Chinese-only brief list inside an otherwise # English page is the tell that a payload was written for one # reader. return { "id": str(e.id), "question_zh": e.question_zh, "question_en": e.question_en, "category": e.category, "views": e.employee_view_count, "helpful": e.employee_helpful_count, "unhelpful": e.employee_unhelpful_count, } return { "visible": int(visible), "hidden": int(hidden), "featured": int(featured), "total_views": int(views or 0), "top_viewed": [_brief(e) for e in top_rows], "needs_rewrite": [_brief(e) for e in unhelpful_rows], "pending_publish": [_brief(e) for e in pending_rows], "at_risk": [_brief(e) for e in at_risk_rows], "gaps": await gap_overview(db), # Semantic-index coverage. An entry with no vector is reachable # only by shared wording, which for a library written in # questionnaire voice means "reachable by luck" — so a shortfall # here is a retrieval outage that no error would ever report. "index": await rag.index_stats(db), }