"""The model's contribution to a person's own access self-attestation. Twice a year every holder is shown their entire access footprint — software seats, cloud grants, permissions, SSO grants, API keys, dataset grants — and asked, per item, "still need this?". The completed campaign is the SOC2 CC6.3 / ISO A.5.15 evidence. The failure mode this addresses is not that people refuse to answer. It is that **the person has nothing to answer with.** A senior engineer's footprint is dozens of rows named ``jira-software-users`` or ``analytics-prod.events_raw``; asked cold, the only safe answer is "keep", and a review where everyone keeps everything produces evidence and changes nothing. PR4 built the missing half: ``asset_seat_usage_signals`` knows, per vendor, when the holder last actually used a product. This module puts that in front of the person deciding — as a **suggestion beside their choice, never in it**. Deliberately the same shape as :mod:`app.hr_lifecycle.mover_review_ai`, including the three properties that make a suggestion safe to show: **It cannot invent a number.** Facts come from the campaign's own items plus the usage signals; ``_validate`` rejects any integer in the narrative that is not one of the published facts. **It cannot invent a category.** Suggestions are matched against the access types actually in this person's footprint; anything else is dropped. **It decides nothing.** The suggestion lands beside the holder's ``decision``, never in it. Nothing is executed from here. One rule this module adds beyond the mover packet, because its facts come from a vendor rather than from Helix's own tables: **A usage signal that is not ``active`` or ``never`` is not evidence of non-use, and neither the prompt nor the validator lets it become one.** ``unknown`` (the vendor declined, or the snapshot went stale) and ``unsupported`` (the vendor cannot answer at all) are published as exactly that. The whole point of carrying four states out of PR4 is lost the moment a summariser flattens them into "5 unused seats". """ from __future__ import annotations import logging import re from collections import Counter from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.core.setting_values import as_bool from app.integrations.models import PlatformSetting logger = logging.getLogger(__name__) _KEY_API = "ai_spending.llm.api_key" _KEY_MODEL = "ai_spending.llm.model" _KEY_REGION = "ai_spending.llm.region" _KEY_ADVICE_ENABLED = "access_review.self_attestation.ai_advice_enabled" #: The model may suggest only these. There is no "transfer" — a #: self-attestation is about the holder's own access, and handing #: something to a named colleague is not a suggestion box's call. REC_KEEP = "keep" REC_RELEASE = "release" _ALLOWED_RECOMMENDATIONS = (REC_KEEP, REC_RELEASE) _MAX_SUMMARY_CHARS = 900 _MAX_REASON_CHARS = 200 #: Below this many days of confirmed non-use, "you have not used it" is #: not an argument worth putting to a holder — people take leave. _IDLE_SUGGESTION_DAYS = 120 @dataclass(slots=True) class AccessTypeFact: access_type: str total: int used_recently: int = 0 idle_confirmed: int = 0 #: Vendor declined / snapshot stale / vendor cannot answer. NOT idle. unknown: int = 0 examples: list[str] = field(default_factory=list) @dataclass(slots=True) class AttestationFacts: #: Carried for the caller's logging / future surfaces only — #: deliberately NOT interpolated into the prompt. The system prompt #: addresses the holder in the second person ("你自己名下的..."), so the #: name adds nothing the model needs, and putting an employee's name #: into a vendor request to make an unused field "used" would be a #: privacy regression dressed up as tidying. holder_display: str total_items: int types: list[AccessTypeFact] #: Types whose usage cannot be measured at all, so the holder is the #: only possible source of truth for them. unmeasurable_types: list[str] = field(default_factory=list) def citable_numbers(self) -> set[int]: numbers = {self.total_items, len(self.types)} for t in self.types: numbers.update({t.total, t.used_recently, t.idle_confirmed, t.unknown}) return numbers async def build_facts( db: AsyncSession, *, items: list[Any], holder_display: str, ) -> AttestationFacts: """Deterministic facts for one holder's footprint. No model involved.""" from app.integrations.usage_signal_service import signals_for_people from app.integrations.usage_signals import SIGNAL_ACTIVE, SIGNAL_NEVER now = datetime.now(UTC) person_ids = sorted({i.person_id for i in items if i.person_id}) signals = await signals_for_people(db, person_ids, now=now) if person_ids else {} by_person_product: dict[tuple[Any, str], dict[str, Any]] = {} sig: dict[str, Any] | None for (person_id, _provider, product), sig in signals.items(): by_person_product[(person_id, product)] = sig totals: Counter[str] = Counter() used: Counter[str] = Counter() idle: Counter[str] = Counter() unknown: Counter[str] = Counter() examples: dict[str, list[str]] = {} for item in items: atype = item.access_type totals[atype] += 1 bucket = examples.setdefault(atype, []) if len(bucket) < 3 and item.access_label: bucket.append(str(item.access_label)[:80]) sig = _signal_for_item(item, by_person_product) if sig is None: unknown[atype] += 1 continue state = sig.get("state") if state == SIGNAL_NEVER: idle[atype] += 1 elif state == SIGNAL_ACTIVE: days = _days_since(sig.get("last_active_at"), now) if days is not None and days >= _IDLE_SUGGESTION_DAYS: idle[atype] += 1 else: used[atype] += 1 else: unknown[atype] += 1 types = [ AccessTypeFact( access_type=atype, total=totals[atype], used_recently=used[atype], idle_confirmed=idle[atype], unknown=unknown[atype], examples=examples.get(atype, []), ) for atype in sorted(totals, key=lambda k: (-totals[k], k)) ] unmeasurable = [t.access_type for t in types if t.total and t.total == t.unknown] return AttestationFacts( holder_display=holder_display, total_items=sum(totals.values()), types=types, unmeasurable_types=unmeasurable, ) def _signal_for_item( item: Any, by_person_product: dict[tuple[Any, str], dict[str, Any]], ) -> dict[str, Any] | None: """Best-effort match from a footprint row to a usage signal. Only software seats can carry one today — cloud grants, permissions, SSO grants, API keys and dataset grants have no vendor last-activity source at all. Returning ``None`` for those is the honest answer and lands them in ``unknown``; inventing a proxy ("granted long ago", "no recent provisioning task") would be exactly the stand-in PR4 refused to build. """ from app.domains.auditor.access_certification_service import AT_SOFTWARE_SEAT if item.access_type != AT_SOFTWARE_SEAT or not item.person_id: return None payload = item.raw_payload or {} product = str(payload.get("usage_product_key") or "").strip() if not product: return None return by_person_product.get((item.person_id, product)) def _days_since(value: Any, now: datetime) -> int | None: if not value: return None if isinstance(value, str): try: value = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None if not isinstance(value, datetime): return None if value.tzinfo is None: value = value.replace(tzinfo=UTC) days: int = (now - value).days return days _SYSTEM = ( "你在帮一位同事看他**自己**名下的访问权限清单,做半年度自查。" "Helix 已经把他持有的东西按类别统计好了,并且在能测到的地方标了使用情况 —— " "你拿到的就是全部依据。\n" "你的任务只有两件:\n" "① 用 3-5 句中文告诉他**这次值得重点看哪几类**、为什么;\n" "② 对每个类别给一个初步建议(keep 或 release),并给出**中英双语**各一句、每句不超过 40 字的理由。\n\n" "硬性约束(违反即整份作废):\n" "- **只能使用给定事实里出现过的数字**。不要相加、不要估算、不要说「大约」。\n" "- **只能对给定的类别作出建议**,不要发明类别。\n" "- 🔴 **「无法确认使用情况」不等于「没在用」。** 事实里标为「无法确认」的条目," "是厂商没告诉我们或者根本查不到,**不是闲置的证据**。绝不可以把这类数量说成" "「闲置」「没用过」「可以清理」;提到它们时只能说「无法确认,需要你自己判断」。\n" "- 你给的是**建议**,不是决定。不要写「已回收」「将删除」这类完成态。\n" "- 拿不准就建议 keep。**误留的代价是下次再问一遍;误撤的代价是他明天上不了班。**\n" "严格输出单个 JSON 对象。" ) def _user_prompt(facts: AttestationFacts) -> str: lines = [ "## 这次自查", f"- 需要确认的条目总数: {facts.total_items}", f"- 涉及类别数: {len(facts.types)}", "", "## 各类别(这是你能引用的全部数字)", ] for t in facts.types: sample = "、".join(t.examples) if t.examples else "(无示例)" lines.append( f"- {t.access_type}: 共 {t.total} 项;" f"其中最近有使用记录 {t.used_recently} 项," f"确认长期未使用 {t.idle_confirmed} 项," f"**无法确认使用情况 {t.unknown} 项**。示例: {sample}" ) if facts.unmeasurable_types: lines += [ "", "## 完全无法测量使用情况的类别", "以下类别 Helix 拿不到任何使用数据,持有人自己是唯一的判断依据 —— " "不要暗示它们闲置:", "- " + "、".join(facts.unmeasurable_types), ] lines += [ "", "## 输出格式", '{"summary_zh": "...", "summary_en": "...", "types": ' '[{"access_type": "...", "recommendation": "keep|release", ' '"reason_zh": "...", "reason_en": "..."}]}', ] return "\n".join(lines) _INT_RE = re.compile(r"\d+") #: Words the model must not attach to an ``unknown`` count. Checked in #: code because the system prompt's ban is a REQUEST, and a request is #: not a gate — this is the one misreading that would turn PR4's #: four-state honesty back into "5 unused seats". _IDLE_WORDS = ( "闲置", "没用过", "未使用", "没在用", "可以清理", "可清理", "unused", "idle", "never used", ) def _validate( payload: dict[str, Any], facts: AttestationFacts, ) -> tuple[str, str, dict[str, tuple[str, str, str]]]: """``(summary_zh, summary_en, {access_type: (rec, reason_zh, reason_en)})``. Raises ``ValueError`` when the reply is ungrounded. The whole narrative is rejected rather than trimmed: a summary with one invented number is not a summary with a small error, it is one nobody can trust the rest of. """ summary_zh = str(payload.get("summary_zh") or "").strip() summary_en = str(payload.get("summary_en") or "").strip() if not summary_zh: raise ValueError("empty summary") allowed = facts.citable_numbers() for text in (summary_zh, summary_en): for token in _INT_RE.findall(text): if len(token) >= 4: continue if int(token) not in allowed: raise ValueError( f"ungrounded number {int(token)} (published: {sorted(allowed)})", ) # Unmeasurable-as-idle check. Only fires when there IS something # unmeasurable to misdescribe, and only when that count is not also a # legitimate idle count — so a fully-measurable footprint is never # rejected for using the word "闲置" correctly. unknown_total = sum(t.unknown for t in facts.types) idle_counts = {t.idle_confirmed for t in facts.types} if unknown_total and unknown_total not in idle_counts: for text in (summary_zh, summary_en): lowered = text.lower() if not any(w in lowered for w in _IDLE_WORDS): continue if any(int(tok) == unknown_total for tok in _INT_RE.findall(text)): raise ValueError( f"described the unmeasurable count ({unknown_total}) as idle", ) known = {t.access_type for t in facts.types} recommendations: dict[str, tuple[str, str, str]] = {} for entry in payload.get("types") or []: if not isinstance(entry, dict): continue atype = str(entry.get("access_type") or "").strip() rec = str(entry.get("recommendation") or "").strip().lower() if atype not in known or rec not in _ALLOWED_RECOMMENDATIONS: continue # Both languages. Most employees on this platform read English, # so a Chinese-only reason next to an English summary is the # majority reader getting the half they cannot use. ``reason`` # (singular) is accepted as a fallback so a model that ignores # the two-key format still yields something rather than nothing. single = str(entry.get("reason") or "").strip() reason_zh = str( entry.get("reason_zh") or single, ).strip()[:_MAX_REASON_CHARS] reason_en = str( entry.get("reason_en") or single, ).strip()[:_MAX_REASON_CHARS] recommendations[atype] = (rec, reason_zh, reason_en) return ( summary_zh[:_MAX_SUMMARY_CHARS], summary_en[:_MAX_SUMMARY_CHARS], recommendations, ) def deterministic_priority(facts: AttestationFacts) -> list[str]: """Access types worth looking at first, with no model involved. What the surfaces fall back to when the model is absent or ungrounded — confirmed idleness first, then volume. A holder who gets no narrative still gets a sensible order rather than a blank space that reads like "nothing to worry about". """ return [ t.access_type for t in sorted( facts.types, key=lambda t: (-t.idle_confirmed, -t.total, t.access_type), ) ] async def _client(db: AsyncSession): from app.domains.bailian.llm_client import BailianLLMClient async def _setting(key: str) -> str | None: row = await db.get(PlatformSetting, key) value = row.value if row else None return value if isinstance(value, str) else None api_key = (await _setting(_KEY_API) or "").strip() if not api_key: return None return BailianLLMClient( api_key=api_key, model=await _setting(_KEY_MODEL), region=await _setting(_KEY_REGION), timeout=45.0, # The LLM call ledger attributes spend by ``purpose``; without # one, this feature's calls book to ``unattributed``. It was the # only construction site in the codebase missing it, which made # "which feature burned the tokens" answerable for 33 callers # and not for the one that fans out once per employee. purpose="auditor.self_attestation_advice", ) # ── failure codes, mirrored from mover_review_ai ───────────────────── FAIL_DISABLED = "disabled" FAIL_NO_CLIENT = "no_client" FAIL_NO_ITEMS = "no_items" FAIL_CALL = "call_failed" FAIL_UNGROUNDED = "ungrounded" async def advice_enabled(db: AsyncSession) -> bool: """Whether holders are shown AI advice at all. Default **off**: this is a new employee-facing surface, and the standing decision for those is per-class opt-in rather than switched on for everyone at once. """ # ``as_bool``, not ``bool``: the admin upsert endpoint takes # ``value: Any``, so this flag can legitimately arrive as the JSON # STRING ``"false"`` — and ``bool("false")`` is True, which would # switch an employee-facing surface on by itself. row = await db.get(PlatformSetting, _KEY_ADVICE_ENABLED) return as_bool(row.value if row is not None else None, default=False) async def annotate_holder( db: AsyncSession, *, certification_id: Any, person_id: Any, items: list[Any], holder_display: str, ) -> Any: """Generate (or refresh) one holder's advice row. Never raises. Every failure path records ``failure_code`` and leaves the campaign exactly as usable as it was — the deterministic :func:`deterministic_priority` ordering is what a holder actually acts on, and the narrative is a convenience on top of it. """ from sqlalchemy import select as _select from app.domains.auditor.models import AuditorSelfAttestationAdvice row = (await db.execute( _select(AuditorSelfAttestationAdvice).where( AuditorSelfAttestationAdvice.certification_id == certification_id, AuditorSelfAttestationAdvice.person_id == person_id, ) )).scalars().first() if row is None: row = AuditorSelfAttestationAdvice( certification_id=certification_id, person_id=person_id, ) db.add(row) facts = await build_facts(db, items=items, holder_display=holder_display) row.facts_snapshot = _facts_payload(facts) row.summary_zh = None row.summary_en = None row.type_recommendations = None row.generated_at = datetime.now(UTC) if not facts.types: row.failure_code = FAIL_NO_ITEMS return row client = await _client(db) if client is None: row.failure_code = FAIL_NO_CLIENT return row try: payload = await client.chat_json( system=_SYSTEM, user=_user_prompt(facts), max_tokens=2000, temperature=0.2, ) except Exception as exc: # noqa: BLE001 — recorded, never raised logger.warning( "self_attestation_ai: call failed for person=%s: %s", person_id, exc, ) row.failure_code = FAIL_CALL return row try: summary_zh, summary_en, recs = _validate(payload or {}, facts) except ValueError as exc: # An ungrounded reply is DISCARDED whole, not trimmed. A summary # with one invented number is not a summary with a small error; # it is one nobody can trust the rest of. logger.warning( "self_attestation_ai: ungrounded reply for person=%s: %s", person_id, exc, ) row.failure_code = FAIL_UNGROUNDED return row row.failure_code = None row.summary_zh = summary_zh row.summary_en = summary_en row.type_recommendations = { atype: { "recommendation": rec, "reason_zh": reason_zh, "reason_en": reason_en, } for atype, (rec, reason_zh, reason_en) in recs.items() } return row def _facts_payload(facts: AttestationFacts) -> dict[str, Any]: """The fact package as stored evidence. Kept so a holder or auditor who disagrees with a suggestion can see what it was told — and so "the model was wrong" can be told apart from "the model was right about stale inputs". """ return { "total_items": facts.total_items, "unmeasurable_types": list(facts.unmeasurable_types), "priority": deterministic_priority(facts), "types": [ { "access_type": t.access_type, "total": t.total, "used_recently": t.used_recently, "idle_confirmed": t.idle_confirmed, "unknown": t.unknown, } for t in facts.types ], } __all__ = [ "FAIL_CALL", "FAIL_DISABLED", "FAIL_NO_CLIENT", "FAIL_NO_ITEMS", "FAIL_UNGROUNDED", "REC_KEEP", "REC_RELEASE", "AccessTypeFact", "AttestationFacts", "advice_enabled", "annotate_holder", "build_facts", "deterministic_priority", ]