"""The security answer library's half of the knowledge loop. The library's defect is precisely diagnosed and already written down, in ``security_faq_rag``'s own module docstring: an employee said 「这个搜索很蠢」, and the reason was not weighting. All 71 entries are phrased as a **customer security questionnaire** — 「贵公司是否部署数据防泄漏(DLP)方案?」 — while an employee asks 「我能用个人电脑写公司代码吗」. Those two sentences share no substring, so lexical search finds nothing. Semantic retrieval closed part of that gap. What it cannot close is **vocabulary**: 「个人电脑」 and 「BYOD」 are not especially close in vector space, and only the second is in the text. Closing it needs somebody to write the employee's words down — and until the loop existed there was nowhere to put them and no queue to review them. So this corpus takes part in the loop with **exactly one kind of proposal**: add the words. Everything else is refused, deliberately. **Why no ``new_article`` here.** A gap with no matching entry means the company has no published position on something. Writing that position is the security team's job, it goes through their review, and it is stated on the company's behalf — a generated draft sitting in a review queue would be the single most dangerous artefact this whole system could produce. That case already has a home: ``auditor_security_faq_gaps`` is the security team's own backlog and they already work it. This module adds nothing to it and takes nothing from it. """ from __future__ import annotations import json import logging import uuid from dataclasses import dataclass from typing import Any from sqlalchemy import case, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.knowledge_loop import service as loop from app.core.knowledge_loop.constants import ( CORPUS_SECURITY_FAQ, KIND_ADD_ALIASES, SOURCE_LINT, ) from app.core.knowledge_loop.registry import ProposalApplyError, register_adapter from app.domains.auditor.models import ( AuditorAnswerLibraryEntry, AuditorSecurityFaqGap, ) logger = logging.getLogger(__name__) #: Gaps examined per pass, most-asked first. Each one costs a retrieval plus a #: model call, and the tail of a gap ledger is mostly one-off phrasings. MAX_GAPS_PER_PASS = 30 def _phrasing_lines(entry: AuditorAnswerLibraryEntry) -> list[str]: return [ line.strip() for line in (entry.employee_phrasings or "").splitlines() if line.strip() ] def merge_phrasings(existing: str | None, additions: list[str]) -> str: """Union, order-preserving, case-insensitively deduplicated.""" lines = [line.strip() for line in (existing or "").splitlines() if line.strip()] seen = {line.casefold() for line in lines} for raw in additions: cleaned = " ".join(str(raw).split()) if cleaned and cleaned.casefold() not in seen: lines.append(cleaned) seen.add(cleaned.casefold()) return "\n".join(lines) class SecurityFaqAdapter: """How an accepted proposal lands on the security answer library.""" corpus = CORPUS_SECURITY_FAQ async def _entry( self, db: AsyncSession, ref: str | None, ) -> AuditorAnswerLibraryEntry: if not ref: raise ProposalApplyError("这条提案没有指明答案库条目。") try: key = uuid.UUID(str(ref)) except (TypeError, ValueError) as exc: raise ProposalApplyError(f"条目 id 不合法:{ref}") from exc entry = await db.get(AuditorAnswerLibraryEntry, key) if entry is None: raise ProposalApplyError( "这条答案库条目已经不存在了。", superseded=True, ) if entry.status != "active": raise ProposalApplyError( "这条答案库条目已停用,提案不再适用。", superseded=True, ) return entry async def describe( self, db: AsyncSession, *, proposal: Any, ) -> dict[str, Any]: """The entry as it stands today, in both stored languages. Both languages, because the reviewer here is deciding whether an employee's phrasing really belongs to this entry, and the entry's own question exists twice — a phrasing that fits the Chinese half and contradicts the English one is a real thing to catch. """ out: dict[str, Any] = {"corpus": self.corpus} if not proposal.target_ref: return out try: entry = await db.get( AuditorAnswerLibraryEntry, uuid.UUID(str(proposal.target_ref)), ) except (TypeError, ValueError): return out if entry is None: out["target"] = {"missing": True, "id": proposal.target_ref} return out out["target"] = { "id": str(entry.id), "title": entry.question_zh, "summary": entry.question_en, # Both answers, so the reviewer sees what the phrasing would be # attached to rather than only what it would be filed under. "body": f"{entry.answer_zh}\n\n---\n\n{entry.answer_en}", "status": entry.status, "category": entry.category, "aliases": _phrasing_lines(entry), "updated_at": str(entry.updated_at or ""), # Nothing in this corpus is machine-owned, so nothing is read-only # for the reason the KB's mirrored pages are. Stated explicitly so # the console's shared drawer does not have to guess. "read_only": False, } return out async def apply( self, db: AsyncSession, *, proposal: Any, actor_person_id: Any, trace_id: str, ) -> dict[str, Any]: if proposal.kind != KIND_ADD_ALIASES: # Refused rather than silently ignored, and the message says where # the work actually belongs. A security answer is stated on the # company's behalf; drafting one is the security team's job and it # goes through their review, not this queue. raise ProposalApplyError( f"安全答案库只接受「补员工问法」这一种提案,不接受 {proposal.kind}。" "缺少某个立场的答案本身,要在答案库管理页由安全团队撰写并复核 ——" "那条路已经有了(安全 FAQ 缺口台账)。", ) entry = await self._entry(db, proposal.target_ref) additions = [ str(item) for item in ((proposal.payload or {}).get("aliases") or []) ] merged = merge_phrasings(entry.employee_phrasings, additions) # 🔴 A no-op is a no-op for the ENTRY, not for the gap ledger. # # Returning early here skipped ``_close_gaps`` and the audit row, and # ``decide`` then marked the proposal ``accepted`` — success reported # while the ledger row this whole path exists to drain stayed open # forever, still counted by ``library_health.open_gaps``. The way in is # ordinary: a security admin pastes the same phrasing onto the entry by # hand before the reviewer gets to the queue. noop = merged == (entry.employee_phrasings or "") if not noop: entry.employee_phrasings = merged # The phrasings are part of the embedding input now, so a stale # vector would keep the library from being found by exactly the # words this change was made to catch. Cleared here rather than in # the migration: clearing all 71 at once would degrade every # question to lexical-only until the re-embed drained, and # lexical-only is the path this whole change exists to stop relying # on. entry.embedding = None entry.embedding_model = None entry.embedded_at = None # The gap that produced this is now answerable. Closing it here rather # than leaving it open is the difference between a ledger that drains # and one that only grows: the security team should see what is still # genuinely unanswered, not a backlog of questions somebody already fixed. closed = await self._close_gaps( db, entry_id=entry.id, phrasings=additions, actor_person_id=actor_person_id, ) from app.audit.service import AuditService await AuditService(db).log( trace_id=trace_id or "knowledge-loop", actor_type="user", actor_user_id=actor_person_id, action="auditor.security_faq.phrasings_added", status="success", resource_type="auditor_answer_library_entry", resource_id=str(entry.id), detail={ "added": additions, "gaps_closed": closed, "already_present": noop, "via": "knowledge_loop", }, ) return { "applied_ref": str(entry.id), "phrasings": len(_phrasing_lines(entry)), "gaps_closed": closed, # Reported rather than hidden: "accepted, and the entry already # said this" is a different outcome from "accepted, and the entry # now says something new", and the reviewer is entitled to know # which one their click produced. "noop": noop, } async def _close_gaps( self, db: AsyncSession, *, entry_id: uuid.UUID, phrasings: list[str], actor_person_id: Any, ) -> int: """Resolve the gap rows whose question this phrasing now answers. Matched on the NORMALISED question the ledger already stores, so the match is the ledger's own notion of "the same question" rather than a second one invented here. """ from app.domains.auditor.security_faq_service import normalize_query wanted = {normalize_query(text) for text in phrasings if text.strip()} wanted.discard("") if not wanted: return 0 rows = ( ( await db.execute( select(AuditorSecurityFaqGap).where( AuditorSecurityFaqGap.status == "open", AuditorSecurityFaqGap.query_norm.in_(sorted(wanted)), ) ) ) .scalars() .all() ) for row in rows: row.status = "resolved" row.resolved_entry_id = entry_id row.resolved_by_person_id = actor_person_id if rows: await db.flush() return len(rows) async def library_health(db: AsyncSession) -> dict[str, Any]: """What the console shows about this corpus's own completeness. ``with_phrasings`` against ``employee_visible`` is the number that matters: an entry employees can see, written in questionnaire register, with nobody's real words attached to it, is an entry they will not find. ``open_gaps`` is the queue of proof. """ from sqlalchemy import func row = ( await db.execute( select( func.count(), func.sum( case( (AuditorAnswerLibraryEntry.employee_visible.is_(True), 1), else_=0, ) ), func.sum( case( ( AuditorAnswerLibraryEntry.employee_phrasings.is_not( None, ), 1, ), else_=0, ) ), ).where(AuditorAnswerLibraryEntry.status == "active") ) ).one() open_gaps = ( await db.execute( select(func.count()) .select_from(AuditorSecurityFaqGap) .where(AuditorSecurityFaqGap.status == "open") ) ).scalar() or 0 return { "entries": int(row[0] or 0), "employee_visible": int(row[1] or 0), "with_phrasings": int(row[2] or 0), "open_gaps": int(open_gaps), } # --------------------------------------------------------------------------- # The lint pass # --------------------------------------------------------------------------- _VOCAB_SYSTEM = """一个员工在安全 FAQ 里这样问,而没有得到答案。给定他的原话和 答案库里**最应该回答这个问题**的那条条目,判断: - 如果那条条目**确实**回答了这个问题,只是措辞对不上 —— relation 填 "vocabulary",并给出应当补进条目「员工问法」的说法。答案库里的条目全部是按 客户安全问卷的口吻写的(「贵公司是否部署…」),而员工是用自己的话问的 (「我能不能…」),所以这种情况很常见。 - 如果那条条目**并没有**回答这个问题 —— relation 填 "not_covered",说法留空。 这是公司还没有这个立场,补问法只会把员工引到一条答不了他的条目上,比搜不到 更糟。**这个判断要严格**:安全问题上,一个自信的错答复是以公司的名义给出的。 给出的说法要用员工的词,每条一句,不要把整段答案塞进去。reason 用中文一句话。""" @dataclass class SecurityLintReport: gaps_examined: int = 0 proposals: int = 0 not_covered: int = 0 llm_available: bool = True def as_dict(self) -> dict[str, Any]: return { "gaps_examined": self.gaps_examined, "proposals": self.proposals, "not_covered": self.not_covered, "llm_available": self.llm_available, } async def lint_security_vocabulary( db: AsyncSession, *, limit: int = MAX_GAPS_PER_PASS, ) -> dict[str, Any]: """Turn the security gap ledger into 'add these words' proposals. Reads the ledger the employee-facing security page already writes — the questions people really asked and did not get answered — and, for each one the library COULD have answered, proposes the phrasing that would have found it. Nothing is proposed for a gap the library genuinely does not cover. Those stay in the ledger, which is where the security team already looks, and which is the only place qualified to decide that the company needs a position it does not have. """ from app.domains.support.copilot import one_shot_json report = SecurityLintReport() gaps = ( ( await db.execute( select(AuditorSecurityFaqGap) .where(AuditorSecurityFaqGap.status == "open") .order_by( AuditorSecurityFaqGap.ask_count.desc(), AuditorSecurityFaqGap.last_asked_at.desc(), ) .limit(max(1, min(limit, 100))) ) ) .scalars() .all() ) if not gaps: return report.as_dict() from app.domains.auditor import security_faq_service as security for gap in gaps: question = (gap.query_sample or "").strip() if len(question) < 3: continue report.gaps_examined += 1 # Retrieval only — no composition, no gap recording. The gap is already # recorded (that is what we are reading), and composing an answer here # would spend a model call on prose nobody reads. found = await security.ask( db, query=question, lang=gap.lang or "zh", person_id=None, limit=1, compose=False, record_gaps=False, ) items = found.get("items") or [] if not items: # Retrieval places it nowhere at all. Not a vocabulary problem — # there is no entry to attach a word to. continue entry_id = str(items[0].get("id") or "") if not entry_id: continue verdict = await one_shot_json( db, _VOCAB_SYSTEM, json.dumps( { "employee_question": question, "asked_times": gap.ask_count, "entry": { "id": entry_id, "question_zh": items[0].get("question"), "answer": (items[0].get("answer") or "")[:1500], }, }, ensure_ascii=False, ) + '\n\n请输出 JSON:{"relation":"vocabulary|not_covered",' '"phrasings":["员工会用的说法", ...],"reason":"中文一句话"}', timeout=40.0, max_tokens=500, ) if verdict is None: # No classifier means the rest would be guesses, and a guess here # attaches an employee's question to an entry that does not answer # it — worse than not finding anything. report.llm_available = False return report.as_dict() if str(verdict.get("relation") or "").strip().lower() != "vocabulary": report.not_covered += 1 continue phrasings = [ " ".join(str(item).split()) for item in (verdict.get("phrasings") or []) if str(item).strip() ][:6] if not phrasings: continue row = await loop.propose( db, corpus=CORPUS_SECURITY_FAQ, kind=KIND_ADD_ALIASES, title=f"给安全条目补 {len(phrasings)} 条员工问法", source=SOURCE_LINT, source_ref=f"security_gap:{gap.id}", target_ref=entry_id, payload={"aliases": phrasings}, rationale=str(verdict.get("reason") or "").strip()[:1000] or None, evidence={ "employee_question": question, "asked_times": gap.ask_count, }, ) if row: report.proposals += 1 return report.as_dict() register_adapter(SecurityFaqAdapter()) __all__ = [ "MAX_GAPS_PER_PASS", "library_health", "SecurityFaqAdapter", "SecurityLintReport", "lint_security_vocabulary", "merge_phrasings", ]