"""Audited recommendation materialization and lifecycle transitions.""" from __future__ import annotations import hashlib import re from collections import defaultdict from collections.abc import Iterable from dataclasses import dataclass from datetime import UTC, datetime, timedelta from decimal import Decimal from typing import Any from uuid import UUID from sqlalchemy import case, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException from app.intelligence.models import ( IntelligenceAnalysis, IntelligenceRecommendation, IntelligenceRecommendationSavingsLink, IntelligenceSnapshot, ) from app.notifications.models import NotificationOutbox from app.notifications.service import NotificationEnvelope, NotificationService #: Statuses a human may set through the HTTP / MCP mutation. _MANUAL_STATUSES = frozenset({"new", "viewed", "in_progress", "resolved", "ignored"}) #: ``superseded`` is machine-written only: re-materialization archives an open #: suggestion the newest analysis stopped emitting (see #: ``_supersede_unseen_recommendations``). It is a persisted status — a stored #: row can carry it, and ``_ALLOWED_TRANSITIONS`` lets a human pull it back #: into the funnel — but no request may move a row INTO it, which is why the #: two sets are separate. _STATUSES = _MANUAL_STATUSES | {"superseded"} #: Still on someone's plate. The one definition of "open"; the read ordering, #: the ROI's ``active`` count and the supersede sweep all key on it. _OPEN_STATUSES = frozenset({"new", "viewed", "in_progress"}) #: Open AND unclaimed — the only rows the supersede sweep may archive. #: ``in_progress`` is excluded deliberately: a human explicitly started that #: one, and ``savings_reconciliation`` links real CMDB savings events to #: in-progress rows, so archiving one would both yank a card out from under #: whoever is working it and orphan its pending attribution. _SWEEPABLE_STATUSES = frozenset({"new", "viewed"}) #: Off the board entirely — never read back into a panel or an ROI total. #: ``resolved`` is deliberately NOT here: a booked outcome keeps counting, it #: just stops competing for attention with open work. _CLOSED_STATUSES = frozenset({"ignored", "superseded"}) #: How many successful analyses (this one included) must fail to raise a #: suggestion before it is archived. Counting RUNS rather than days is what #: makes this survive a schedule change or a stalled scheduler: 14 days of #: silence means nothing if the job has not run, and everything if it has run #: daily. _SUPERSEDE_AFTER_MISSED_RUNS = 2 #: …and a floor in wall-clock time, because runs are also human-triggerable. #: Three force-generates in an afternoon are three runs; without this an #: operator clicking 「生成简报」 a few times would clear the board in a day. #: Under the shipped weekly cadence (migration 522, Mondays 07:35 UTC) the run #: count is the binding constraint and this floor never is. _SUPERSEDE_MIN_AGE_DAYS = 7 #: Sort key that keeps open suggestions ahead of handled ones. _OPEN_FIRST = case( (IntelligenceRecommendation.status.in_(sorted(_OPEN_STATUSES)), 0), else_=1, ) _PRIORITY = {"p0": 0, "p1": 1, "p2": 2, "p3": 3} _SEVERITY = {"p0": "critical", "p1": "high", "p2": "medium", "p3": "low"} _ALL_DIMENSIONS = frozenset({ "people_org", "lifecycle", "assets", "software_api", "cloud_network", "cost_efficiency", "security_compliance", "platform_ops", }) # Where a manager is sent when they start handling a recommendation. # # EVERY value in this module must be a declared desktop SPA route. Five of the # eight dimension entries used to name paths that are not routes on either # shell (``/admin/people``, ``/admin/software``, ``/admin/cloud``, # ``/admin/cost``, ``/admin/security``), so "开始处理" navigated the manager to # the 404 page — on the phone it also hard-left the Feishu H5 shell with no way # back. ``tests/test_intelligence/test_recommendation_handoff.py`` now parses # ``web/src/App.tsx`` and fails on any path that is not a real route, and # ``tests/test_mobile/test_deep_link_contract.py`` fails on any path that is # not classified in ``contracts/mobile-deep-links.json``. # # Only ``global_admin`` can start a handoff (``can_manage_recommendations``), # so these targets do not need to be role-narrowed: the one role that follows # them can open all of them. _WORKBENCH_BY_DIMENSION = { "people_org": "/admin/identity/data-quality", "lifecycle": "/admin/lifecycle", "assets": "/admin/assets", "software_api": "/admin/assets/software-overview", "cloud_network": "/admin/cloud/overview", "cost_efficiency": "/admin/finance/overview", "security_compliance": "/admin/security-insights", "platform_ops": "/admin/health", } # Refinement keyed on the leaf of a GROUNDED fact citation (see # ``_fact_address`` for the three spellings that reach us), so the handoff # opens the surface where that specific deduction is actually fixed instead of # the domain's front door. # A recommendation about ``hris_missed_cycles`` belongs on the HRIS sync # console, not on a generic people page. # # Keys are the real ``aggregate_facts`` names the providers publish; an entry # whose leaf no provider emits would be a silent no-op, so # ``test_recommendation_handoff.py`` asserts every key exists in some # provider's fact package. _WORKBENCH_BY_FACT: dict[tuple[str, str], str] = { # people_org — org/manager/prediction gaps are listed as data-quality # findings; freshness is fixed at the HRIS inbound console. ("people_org", "hris_freshness_state"): "/admin/hr/sync", ("people_org", "hris_missed_cycles"): "/admin/hr/sync", # lifecycle ("lifecycle", "day1_total_count"): "/admin/onboarding/cockpit", ("lifecycle", "day1_ready_count"): "/admin/onboarding/cockpit", ("lifecycle", "open_past_due_hard_gap_count"): "/admin/onboarding/cockpit", ("lifecycle", "onboarding_sla_total_count"): "/admin/onboarding/cockpit", ("lifecycle", "onboarding_sla_ok_count"): "/admin/onboarding/cockpit", ("lifecycle", "leaver_total_count"): "/admin/offboarding", ("lifecycle", "leaver_failed_count"): "/admin/offboarding", # assets ("assets", "inventory_item_count"): "/admin/assets/inventory", ("assets", "inventory_confirmed_count"): "/admin/assets/inventory", ("assets", "inventory_issue_count"): "/admin/assets/inventory", ("assets", "long_idle_count"): "/admin/assets/list", ("assets", "leaver_held_count"): "/admin/assets/list", ("assets", "optimization"): "/admin/assets/list", ("assets", "warranty_population_count"): "/admin/assets/financial", ("assets", "warranty_complete_count"): "/admin/assets/financial", ("assets", "financial_complete_count"): "/admin/assets/financial", # software_api ("software_api", "live_key_count"): "/admin/assets/api-keys", ("software_api", "expiry_governed_key_count"): "/admin/assets/api-keys", ("software_api", "overdue_rotation_count"): "/admin/assets/api-keys", ("software_api", "vendor_holder_count"): "/admin/assets/seat-allocations", ("software_api", "matched_holder_count"): "/admin/assets/seat-allocations", ("software_api", "open_reconciliation_anomaly_count"): "/admin/assets/seat-allocations", # cloud_network ("cloud_network", "stale_scope_count"): "/admin/cloud/sync-jobs", ("cloud_network", "sync_missed_cycles"): "/admin/cloud/sync-jobs", ("cloud_network", "active_grant_count"): "/admin/cloud/iam-users", ("cloud_network", "governed_grant_count"): "/admin/cloud/iam-users", ("cloud_network", "active_domain_count"): "/admin/cloud/certificates", ("cloud_network", "certificate_count"): "/admin/cloud/certificates", ("cloud_network", "actionable_efficiency_count"): "/admin/cloud/efficiency", ("cloud_network", "owner_object_count"): "/admin/cloud/scopes", ("cloud_network", "owner_covered_count"): "/admin/cloud/scopes", # cost_efficiency ("cost_efficiency", "optimizable_resource_count"): "/admin/cloud/efficiency", ("cost_efficiency", "priced_efficiency_resource_count"): "/admin/cloud/efficiency", ("cost_efficiency", "estimated_monthly_savings_usd"): "/admin/cloud/efficiency", ("cost_efficiency", "anomalous_resource_count"): "/admin/cloud/efficiency", ("cost_efficiency", "business_budget_scope_count"): "/admin/finance/budgets", ("cost_efficiency", "allocated_business_cost_usd"): "/admin/finance/cost-centers", # security_compliance ("security_compliance", "mfa_observation_count"): "/auditor/mfa", ("security_compliance", "mfa_compliant_count"): "/auditor/mfa", ("security_compliance", "privileged_review_item_count"): "/compliance/controls", ("security_compliance", "privileged_review_decided_count"): "/compliance/controls", ("security_compliance", "exposure_finding_count"): "/auditor/public-exposure", ("security_compliance", "open_exposure_count"): "/auditor/public-exposure", ("security_compliance", "credential_observation_count"): "/auditor/secret-rotation", ("security_compliance", "credential_risk_count"): "/auditor/secret-rotation", ("security_compliance", "evidence_job_count"): "/auditor/evidence-packs", ("security_compliance", "evidence_observed_job_count"): "/auditor/evidence-packs", ("security_compliance", "evidence_missing_job_count"): "/auditor/evidence-packs", ("security_compliance", "evidence_missed_cycles"): "/auditor/evidence-packs", # platform_ops ("platform_ops", "enabled_job_count"): "/admin/scheduled-jobs", ("platform_ops", "observed_job_count"): "/admin/scheduled-jobs", ("platform_ops", "terminal_job_run_count_7d"): "/admin/scheduled-jobs", ("platform_ops", "failed_job_run_count_7d"): "/admin/scheduled-jobs", ("platform_ops", "eligible_automatic_provisioning_count_30d"): "/admin/provisioning/tasks", ("platform_ops", "evaluable_automatic_provisioning_count_30d"): "/admin/provisioning/tasks", ("platform_ops", "failed_automatic_provisioning_count_30d"): "/admin/provisioning/tasks", ("platform_ops", "notification_count_30d"): "/admin/notifications/outbox", ("platform_ops", "terminal_notification_count_30d"): "/admin/notifications/outbox", ("platform_ops", "notification_failure_count_30d"): "/admin/notifications/outbox", ("platform_ops", "enabled_connector_count"): "/admin/integrations", ("platform_ops", "healthy_connector_count"): "/admin/integrations", ("platform_ops", "open_data_quality_finding_count"): "/admin/identity/data-quality", ("platform_ops", "high_data_quality_finding_count"): "/admin/identity/data-quality", } # A recommendation grounded in more than one dimension has no single workbench. # The intelligence center is the honest destination — it is where the evidence # for every cited dimension is readable side by side — and it now has a mobile # twin, so the phone stays in H5. _CROSS_DOMAIN_WORKBENCH = "/admin/intelligence" # ``assign`` is gated deliberately: handing a resource a new owner moves # accountability, which is the same kind of change as ``transfer``. Erring # toward the gate can only add friction; erring the other way removes one. _APPROVAL_ACTIONS = frozenset( {"purchase", "approve", "configure", "transfer", "assign"} ) # These symbolic steps are deliberately rendered by the client from its own # locale catalog. They are not model prose and, unlike a URL or mutation # payload, cannot turn a recommendation into an arbitrary write primitive. _HANDLING_STEPS_BY_ACTION = { "investigate": ("review_evidence", "open_workbench", "verify_snapshot"), "reclaim": ("review_evidence", "open_workbench", "verify_snapshot"), "remediate": ("review_evidence", "open_workbench", "verify_snapshot"), "configure": ("review_evidence", "approval_boundary", "open_workbench", "verify_snapshot"), "purchase": ("review_evidence", "approval_boundary", "open_workbench", "verify_snapshot"), "approve": ("review_evidence", "approval_boundary", "open_workbench", "verify_snapshot"), "transfer": ("review_evidence", "approval_boundary", "open_workbench", "verify_snapshot"), "notify": ("review_evidence", "send_notification", "verify_snapshot"), "assign": ("review_evidence", "approval_boundary", "open_workbench", "verify_snapshot"), } def _fact_address(ref: str) -> tuple[str, str] | None: """Split a citation into ``(dimension, leaf)`` whichever way it is spelled. Three spellings are live at once and all of them are legitimate: * ``/facts/assets/long_idle_count`` — the nested JSON Pointer, which is what contracts persisted before the fact package was flattened; * ``/facts/assets.long_idle_count`` — the pointer prefix in front of a flattened key; * ``assets.long_idle_count`` — the bare flattened key, which the contract began accepting once flattening made the keys look like whole paths. Reading only the first spelling is not a cosmetic gap: after flattening, real model output stopped matching it, so every recommendation resolved to "no grounded dimension" and fell back to the cross-domain workbench. The per-dimension handoff would have been silently dead in production. """ tokens = [token for token in re.split(r"[/.]", str(ref)) if token] if tokens and tokens[0] == "facts": tokens = tokens[1:] if len(tokens) < 2 or tokens[0] not in _WORKBENCH_BY_DIMENSION: return None return tokens[0], tokens[1] def fact_dimensions(fact_refs: Iterable[str]) -> set[str]: """Every known dimension a set of citations grounds in. THE one place this is decided. Three call sites used to re-implement it inline by splitting on "/" and demanding ``parts[1] == "facts"``, which stopped matching real output the moment the fact package was flattened — and each of those call sites is a VISIBILITY filter, so the failure was silent and total: a role scoped to fewer than all eight dimensions matched nothing and saw an empty recommendation list and a zero ROI. """ return { address[0] for ref in fact_refs if (address := _fact_address(ref)) is not None } def _grounded_dimensions(fact_refs: list[str]) -> list[str]: """Return only known dimensions from local fact pointers.""" return sorted(fact_dimensions(fact_refs)) def _fact_leaves(fact_refs: list[str], dimension: str) -> list[str]: """Leaf fact names cited inside one dimension, in citation order.""" return [ address[1] for ref in fact_refs if (address := _fact_address(ref)) is not None and address[0] == dimension ] def _resolve_workbench_path(fact_refs: list[str], dimension: str) -> str: """Pick the single surface where this recommendation is actually handled. The winner is the surface the most cited facts point at, with a lexicographic tie-break, so the same stored contract always resolves to the same path — a manager who reopens a recommendation must not be sent somewhere else than the first time. """ counts: dict[str, int] = defaultdict(int) for leaf in _fact_leaves(fact_refs, dimension): refined = _WORKBENCH_BY_FACT.get((dimension, leaf)) if refined is not None: counts[refined] += 1 if not counts: return _WORKBENCH_BY_DIMENSION[dimension] return min(counts, key=lambda path: (-counts[path], path)) def build_recommendation_handoff( *, action_kind: str, fact_refs: list[str], snapshot_facts_hash: str | None = None, ) -> dict[str, Any]: """Build a local, auditable handoff to an existing domain boundary. The model may select an allowed action *category*, but never supplies a URL, API operation, approver, or write payload. Those are derived here from grounded fact pointers, preventing a narrative from becoming an arbitrary write or redirect primitive. """ dimensions = _grounded_dimensions(fact_refs) # Every candidate was filtered through ``_WORKBENCH_BY_DIMENSION`` above, so a # single unambiguous dimension always resolves; anything else stays generic. workbench_path = ( _resolve_workbench_path(fact_refs, dimensions[0]) if len(dimensions) == 1 else _CROSS_DOMAIN_WORKBENCH ) return { "contract_version": 3, "action_kind": action_kind, "handoff_type": "existing_workbench", "workbench_path": workbench_path, "requires_approval": action_kind in _APPROVAL_ACTIONS, "target_scope": "single_domain" if len(dimensions) == 1 else "cross_domain", "dimensions": dimensions, "handling_steps": list(_HANDLING_STEPS_BY_ACTION.get( action_kind, ("review_evidence", "open_workbench", "verify_snapshot"), )), "fact_refs": fact_refs, "verification": { "method": "fresh_snapshot", "baseline_facts_hash": snapshot_facts_hash, "success_state": "facts_changed_or_risk_resolved", }, } def upgrade_recommendation_handoff( contract: dict[str, Any] | None, ) -> dict[str, Any]: """Safely upgrade a persisted pre-v3 contract before it is executed. Older recommendations contain only grounded fact references and an action category, which are enough to deterministically rebuild the new display plan. We intentionally do not retain a legacy model-controlled path. v2 rows are re-derived too: their stored ``workbench_path`` was computed from a table in which five of eight dimensions named a path that is not a route, so keeping it would preserve a dead end for every recommendation that already exists. """ current = contract or {} if int(current.get("contract_version", 0) or 0) >= 3: return current return build_recommendation_handoff( action_kind=str(current.get("action_kind") or "investigate").strip().lower(), fact_refs=[str(ref) for ref in current.get("fact_refs", []) if isinstance(ref, str)], snapshot_facts_hash=(current.get("verification") or {}).get("baseline_facts_hash"), ) def recommendation_verification_state( *, baseline_facts_hash: str | None, latest_facts_hash: str | None, latest_data_as_of: datetime | None, ) -> dict[str, str | None]: """Describe a recheck without claiming that a risk or saving is resolved. A different immutable snapshot proves only that the evidence changed. It must remain a human/domain decision whether that change actually resolves the recommendation, so this result deliberately never changes lifecycle status or ROI values. """ if not baseline_facts_hash: state = "baseline_unavailable" elif not latest_facts_hash: state = "awaiting_snapshot" elif latest_facts_hash == baseline_facts_hash: state = "awaiting_change" else: state = "facts_changed_review_required" return { "state": state, "latest_data_as_of": latest_data_as_of.isoformat() if latest_data_as_of else None, } async def _latest_snapshots_by_subject( db: AsyncSession, subject_keys: set[tuple[str, str, str]], ) -> dict[tuple[str, str, str], IntelligenceSnapshot]: """Batch latest immutable snapshots for recommendation verification.""" if not subject_keys: return {} scope_clause = or_(*( (IntelligenceSnapshot.subject_type == subject_type) & (IntelligenceSnapshot.subject_id == subject_id) & (IntelligenceSnapshot.audience_key == audience_key) for subject_type, subject_id, audience_key in subject_keys )) rows = (await db.execute( select(IntelligenceSnapshot) .where(scope_clause) .order_by(IntelligenceSnapshot.created_at.desc()) )).scalars().all() latest: dict[tuple[str, str, str], IntelligenceSnapshot] = {} for row in rows: latest.setdefault((row.subject_type, row.subject_id, row.audience_key), row) return latest @dataclass(frozen=True) class MaterializationResult: """What one re-materialization did to the stored suggestion set. ``created`` alone used to be the whole return value, which is why the panel could only ever grow: a re-emitted suggestion was invisible in every job result, and an abandoned one was invisible everywhere. """ created: int = 0 refreshed: int = 0 superseded: int = 0 async def _supersede_unseen_recommendations( db: AsyncSession, analysis: IntelligenceAnalysis, *, run_at: datetime, ) -> list[IntelligenceRecommendation]: """Archive open suggestions the newest analyses stopped emitting. Without this, ``intelligence_recommendations`` was append-only in practice: nothing aged a row out, nothing superseded it, and the only way off the panel was a human clicking 忽略. A suggestion whose underlying finding disappeared — or whose title the model reworded, which mints a NEW row under the dedupe key — sat in 优先建议 forever. This is not a hypothetical: on production every single stored row has ``first_seen_at == last_seen_at``. The dedupe key ends in ``title_en``, and across the three successful briefs (2026-08-18 / 08-25 / 08-29) the model reworded every title it re-raised — 「回收闲置资产以优化成本」 became 「回收闲置资产」, 「实施MFA和特权访问审查」 became 「启动特权访问评审」 — so the match has never fired once and each run simply added rows. Deliberately narrow: * only OPEN, UNCLAIMED rows (``new`` / ``viewed``). A ``resolved`` row carries a booked outcome (and possibly attested savings) that must keep counting in the ROI totals, an ``ignored`` row already records a human decision, and an ``in_progress`` row is somebody's live work. * only rows that ``_SUPERSEDE_AFTER_MISSED_RUNS`` successful analyses in a row declined to raise — counted in RUNS, not days, so one terse run cannot clear the board and a stalled scheduler cannot let dead rows accumulate forever — AND that are at least ``_SUPERSEDE_MIN_AGE_DAYS`` old, so a human force-generating a few times in an afternoon does not turn three clicks into three "missed" runs. * only within THIS analysis's subject/audience triple, so a platform brief never touches a regional asset audience's suggestions. The autoflush on this SELECT is load-bearing: the caller has already bumped ``last_seen_at`` on every row the current run re-emitted, and those pending UPDATEs must reach the database before the cutoff is evaluated or a suggestion re-stated this morning would be archived. """ # The Nth-newest successful analysis for this subject: anything last seen # before it has been passed over by N runs. ``generated_at`` (not # ``created_at``) because a cache-reused run returns the analysis that # actually produced the text. # ``is not None`` twice on purpose: the SQL already filters it, but # the invariant is what the arithmetic below depends on, and a filter # in another statement is not something the next reader (or the type # checker) can see from here. run_stamps = list(stamp for stamp in (await db.execute( select(IntelligenceAnalysis.generated_at) .where( IntelligenceAnalysis.subject_type == analysis.subject_type, IntelligenceAnalysis.subject_id == analysis.subject_id, IntelligenceAnalysis.audience_key == analysis.audience_key, IntelligenceAnalysis.status == "succeeded", IntelligenceAnalysis.generated_at.is_not(None), ) .order_by(IntelligenceAnalysis.generated_at.desc()) .limit(_SUPERSEDE_AFTER_MISSED_RUNS) )).scalars().all() if stamp is not None) if len(run_stamps) < _SUPERSEDE_AFTER_MISSED_RUNS: # Not enough history to have missed anything yet. return [] # MySQL/TiDB hand back naive datetimes even for a ``timezone=True`` # column, and ``min()`` on a naive/aware pair raises. Everything Helix # stores here is UTC. nth_newest_run = run_stamps[-1] if nth_newest_run.tzinfo is None: nth_newest_run = nth_newest_run.replace(tzinfo=UTC) cutoff = min(nth_newest_run, run_at - timedelta(days=_SUPERSEDE_MIN_AGE_DAYS)) stale = list((await db.execute( select(IntelligenceRecommendation).where( IntelligenceRecommendation.subject_type == analysis.subject_type, IntelligenceRecommendation.subject_id == analysis.subject_id, IntelligenceRecommendation.audience_key == analysis.audience_key, IntelligenceRecommendation.status.in_(sorted(_SWEEPABLE_STATUSES)), IntelligenceRecommendation.last_seen_at < cutoff, ) )).scalars().all()) for row in stale: row.status = "superseded" return stale async def materialize_analysis_recommendations( db: AsyncSession, analysis: IntelligenceAnalysis, ) -> MaterializationResult: """Upsert only validated model recommendations; never create duplicates.""" rows = analysis.recommendations or [] snapshot_id = analysis.snapshot_id run_at = datetime.now(UTC) created = 0 refreshed = 0 for item in rows: if not isinstance(item, dict): continue priority = str(item.get("priority", "p3")) action = str(item.get("action_kind", "investigate")) title_en = str(item.get("title_en", "")) dedupe = hashlib.sha256( f"{analysis.subject_type}:{analysis.subject_id}:{analysis.audience_key}:" f"{priority}:{action}:{title_en}".encode() ).hexdigest() existing = (await db.execute( select(IntelligenceRecommendation).where( IntelligenceRecommendation.subject_type == analysis.subject_type, IntelligenceRecommendation.subject_id == analysis.subject_id, IntelligenceRecommendation.audience_key == analysis.audience_key, IntelligenceRecommendation.priority == _PRIORITY.get(priority, 3), IntelligenceRecommendation.domain == action, IntelligenceRecommendation.title_en == title_en, ).order_by(IntelligenceRecommendation.last_seen_at.desc()).limit(1) )).scalar_one_or_none() if existing is not None: existing.last_seen_at = run_at existing.analysis_id = analysis.id existing.snapshot_id = snapshot_id existing.dedupe_key = dedupe # The prose is re-stated, not frozen at first sight. Only # ``title_en`` participates in the match above, so everything else # the model just wrote about the SAME finding — the Chinese title # and both summaries, which is where the numbers live — used to be # discarded silently. Production carried a cloud-network card # reading "6737 个云网络对象中仅有 247 个有所有者" for weeks after # #3043 fixed that very metric to count inherited subscription # owners (6486 of 6503 servers resolve to one); the row was # re-materialized every week and never once re-read. # All four move together, never field by field: they describe ONE # run's reading of this finding, and a title from Monday over a # summary from three weeks ago is a worse lie than either alone. # No empty-value guard is needed — ``GroundedNarrative`` pins # title/detail at ``min_length=1`` before the analysis is stored. existing.title_zh = str(item.get("title_zh", "")) existing.summary_zh = str(item.get("detail_zh", "")) existing.summary_en = str(item.get("detail_en", "")) existing.evidence_refs = [ {"ref": ref} for ref in item.get("evidence_refs", []) ] existing.action_contract = build_recommendation_handoff( action_kind=action, fact_refs=list(item.get("fact_refs", [])), snapshot_facts_hash=analysis.facts_hash, ) refreshed += 1 continue db.add(IntelligenceRecommendation( snapshot_id=snapshot_id, analysis_id=analysis.id, subject_type=analysis.subject_type, subject_id=analysis.subject_id, audience_key=analysis.audience_key, dedupe_key=dedupe, domain=action, severity=_SEVERITY.get(priority, "low"), priority=_PRIORITY.get(priority, 3), title_zh=str(item.get("title_zh", "")), title_en=title_en, summary_zh=str(item.get("detail_zh", "")), summary_en=str(item.get("detail_en", "")), evidence_refs=[{"ref": ref} for ref in item.get("evidence_refs", [])], action_contract=build_recommendation_handoff( action_kind=action, fact_refs=list(item.get("fact_refs", [])), snapshot_facts_hash=analysis.facts_hash, ), status="new", )) created += 1 # An analysis that published nothing is not evidence that the standing # suggestions are gone — it is far more likely a terse or degraded run — # so a silent run never sweeps. stale = ( await _supersede_unseen_recommendations(db, analysis, run_at=run_at) if rows else [] ) if stale: await AuditService(db).log( trace_id=f"intelligence:recommendation.supersede:{run_at.isoformat()}", actor_type="system", actor_user_id=None, action="intelligence.recommendation.supersede", resource_type="intelligence.recommendation", resource_id=None, status="success", detail={ "analysis_id": str(analysis.id), "subject_type": analysis.subject_type, "subject_id": analysis.subject_id, "audience_key": analysis.audience_key, "missed_runs": _SUPERSEDE_AFTER_MISSED_RUNS, "min_age_days": _SUPERSEDE_MIN_AGE_DAYS, "count": len(stale), # Bounded on purpose: the audit detail is capped at 16 KiB and # the count above is the number that matters. "recommendations": [ {"id": str(row.id), "title_en": row.title_en} for row in stale[:50] ], }, ) await db.commit() return MaterializationResult( created=created, refreshed=refreshed, superseded=len(stale), ) async def read_recommendations( db: AsyncSession, *, subject_keys: set[tuple[str, str, str]], allowed_dimensions: set[str] | None = None, limit: int = 50, ) -> list[dict[str, Any]]: if not subject_keys: return [] scope_clause = or_(*( (IntelligenceRecommendation.subject_type == subject_type) & (IntelligenceRecommendation.subject_id == subject_id) & (IntelligenceRecommendation.audience_key == audience_key) for subject_type, subject_id, audience_key in subject_keys )) rows = (await db.execute( select(IntelligenceRecommendation) .where( IntelligenceRecommendation.status.notin_(sorted(_CLOSED_STATUSES)), scope_clause, ) # Open work first, THEN priority. A resolved P0 is not a P0 any more, # but it used to outrank every open P1 in this list and — because the # window is capped at 200 rows and the caller then takes the first # ``limit`` — a long enough tail of booked outcomes could push open # suggestions out of the response entirely. .order_by( _OPEN_FIRST, IntelligenceRecommendation.priority, IntelligenceRecommendation.created_at.desc(), ) .limit(200) )).scalars().all() latest_by_subject = await _latest_snapshots_by_subject(db, subject_keys) visible = [] for row in rows: if allowed_dimensions is not None and allowed_dimensions != _ALL_DIMENSIONS: refs = (row.action_contract or {}).get("fact_refs", []) if not (fact_dimensions(refs) & allowed_dimensions): continue # A read may project an older contract, but never writes it. This # gives historical recommendations the same explanation immediately; # the upgrade is persisted only when the manager explicitly starts # the existing handoff. contract = upgrade_recommendation_handoff(row.action_contract) latest = latest_by_subject.get((row.subject_type, row.subject_id, row.audience_key)) verification = recommendation_verification_state( baseline_facts_hash=(contract.get("verification") or {}).get("baseline_facts_hash"), latest_facts_hash=latest.facts_hash if latest is not None else None, latest_data_as_of=latest.data_as_of if latest is not None else None, ) visible.append({ "id": str(row.id), "domain": row.domain, "severity": row.severity, "priority": row.priority, "title_zh": row.title_zh, "title_en": row.title_en, "summary_zh": row.summary_zh, "summary_en": row.summary_en, "status": row.status, "evidence_refs": row.evidence_refs, "expected_monthly_savings": str(row.expected_monthly_savings) if row.expected_monthly_savings is not None else None, "action_contract": contract, "verification_state": verification, "created_at": row.created_at.isoformat(), }) return visible[:max(1, min(limit, 200))] async def read_recommendation_roi( db: AsyncSession, *, subject_keys: set[tuple[str, str, str]], allowed_dimensions: set[str] | None = None, ) -> dict[str, Any]: """Return audited outcome totals; estimates are never called realized.""" if not subject_keys: return { "total": 0, "active": 0, "resolved": 0, "expected_monthly_savings": "0", "confirmed_savings": "0", "realized_savings": "0", "manual_realized_savings": "0", "automatic_realized_savings": "0", "automatic_realized_event_count": 0, "model_calls": 0, } scope_clause = or_(*( (IntelligenceRecommendation.subject_type == subject_type) & (IntelligenceRecommendation.subject_id == subject_id) & (IntelligenceRecommendation.audience_key == audience_key) for subject_type, subject_id, audience_key in subject_keys )) rows = (await db.execute(select(IntelligenceRecommendation).where( scope_clause, IntelligenceRecommendation.status.notin_(sorted(_CLOSED_STATUSES)), ))).scalars().all() if allowed_dimensions is not None and allowed_dimensions != _ALL_DIMENSIONS: rows = [ row for row in rows if fact_dimensions( (row.action_contract or {}).get("fact_refs", []) ) & allowed_dimensions ] active = sum(row.status in _OPEN_STATUSES for row in rows) resolved = sum(row.status == "resolved" for row in rows) def total(field: str) -> str: return str(sum((getattr(row, field) or 0) for row in rows)) linked_events = [] if rows: from app.domains.cmdb.models import CmdbEfficiencySavingsEvent linked_events = list((await db.execute( select(CmdbEfficiencySavingsEvent).join( IntelligenceRecommendationSavingsLink, IntelligenceRecommendationSavingsLink.savings_event_id == CmdbEfficiencySavingsEvent.id, ).where(IntelligenceRecommendationSavingsLink.recommendation_id.in_( [row.id for row in rows] )) )).scalars().all()) automatic_realized = Decimal("0") if linked_events: from app.domains.cmdb.savings_service import realized_savings_by_event automatic_realized = sum( (await realized_savings_by_event(db, linked_events)).values(), Decimal("0"), ) manual_realized = sum((row.realized_savings or 0) for row in rows) return { "total": len(rows), "active": active, "resolved": resolved, "expected_monthly_savings": total("expected_monthly_savings"), "confirmed_savings": total("confirmed_savings"), "realized_savings": str(manual_realized + automatic_realized), "manual_realized_savings": str(manual_realized), "automatic_realized_savings": str(automatic_realized), "automatic_realized_event_count": len(linked_events), "model_calls": 0, } async def send_recommendation_notifications( db: AsyncSession, row: IntelligenceRecommendation, *, actor_id: UUID, ) -> int: """Enqueue one deduplicated outbox card per active IT administrator. Delivery remains delegated to the existing outbox retry worker; this mutation only creates durable, auditable pending rows. """ # Lock the immutable recommendation before reading its existing fan-out. # A UI click, MCP call and retrying proxy can otherwise all observe an # empty outbox at once and create duplicate Feishu/email cards. The # parent row is the natural, stable serialisation key: unlike a broad # outbox unique constraint it does not prevent legitimate future alerts # about another occurrence of the same resource. await db.execute( select(IntelligenceRecommendation.id) .where(IntelligenceRecommendation.id == row.id) .with_for_update(), ) # Resolve locale before constructing the string-only notification envelope. from app.core.locale import resolve_lang from app.scheduled_jobs.failure_notifier import list_admin_recipients recipients = await list_admin_recipients(db) if not recipients: return 0 existing = set((await db.execute( select(NotificationOutbox.recipient_person_id).where( NotificationOutbox.related_resource_type == "intelligence.recommendation", NotificationOutbox.related_resource_id == row.id, NotificationOutbox.status != "dead", ) )).scalars().all()) notification = NotificationService(db) # The recommendation's own copy travels as a locale-fragment triple so the # ``intelligence.recommendation`` template can wrap it in the reader's # language (the wrapper prose — severity line, "mark it resolved" — used to # be absent entirely; the envelope carried nothing but the raw model text). # The model produces zh + en only, so the ja slot is filled with English: # that matches ``FALLBACK_CHAIN``'s ja → en → zh order, i.e. a Japanese # reader gets a Japanese frame around English content rather than Chinese. fragments = { "rec_title_zh": row.title_zh, "rec_title_en": row.title_en, "rec_title_ja": row.title_en, "rec_summary_zh": row.summary_zh, "rec_summary_en": row.summary_en, "rec_summary_ja": row.summary_en, } queued = 0 for recipient in recipients: if recipient.id in existing: continue lang = resolve_lang(recipient) title = row.title_zh if lang == "zh" else row.title_en await notification.enqueue(NotificationEnvelope( recipient=recipient, # ``title`` is the no-template fallback and the stored outbox # title; the bell and both channels re-render it from the # template subject per recipient. title=title, body_text="(see template)", channels=["feishu", "email"], template_key="intelligence.recommendation", url=f"/admin/intelligence?recommendation={row.id}", related_resource_type="intelligence.recommendation", related_resource_id=row.id, context={ "actor_id": str(actor_id), "severity": row.severity, **fragments, }, )) queued += 1 await db.commit() return queued #: The only lifecycle moves the product actually offers, made explicit. #: #: Before this map every status was reachable from every other one, so a #: caller could regress ``resolved -> new`` (erasing the fact that the ROI #: was already booked), promote ``ignored -> resolved`` (claiming a saving #: on a suggestion someone explicitly dismissed, with the dismissal reason #: still attached), or jump ``new -> resolved`` and attach confirmed / #: realized savings to a row nobody ever opened. None of those are reachable #: from the desktop panel or the H5 sheet — both only offer #: ``new -> viewed``, ``new|viewed -> in_progress`` (start handoff) and #: ``in_progress -> resolved`` — so the guard codifies the shipped flow #: rather than inventing a new one. #: #: Re-entering the same status is always allowed: rule 5 (every #: state-changing operation is idempotent) means a retried PATCH or MCP call #: must not 422. ``resolved -> resolved`` is also the path a savings #: correction takes, which is why ``resolved_at`` is written once only. _ALLOWED_TRANSITIONS: dict[str, frozenset[str]] = { # A fresh suggestion is read, started or dismissed — never resolved # before anybody handled it. "new": frozenset({"viewed", "in_progress", "ignored"}), "viewed": frozenset({"in_progress", "resolved", "ignored"}), "in_progress": frozenset({"viewed", "resolved", "ignored"}), # Reopening a booked outcome is legitimate (the fix regressed); silently # rewinding it to an unhandled state is not. "resolved": frozenset({"in_progress"}), # A dismissal can be taken back into the funnel, but must be re-handled # before it can claim an outcome. "ignored": frozenset({"new", "viewed", "in_progress"}), # Archived because the analysis stopped emitting it. Nobody decided # anything about it, so a human may still take it back — same shape as a # dismissal. The entry is also what keeps a PATCH against an archived row # a 422 instead of a KeyError 500: this dict is indexed by the CURRENT # status, so a persisted status missing from it crashes the mutation. "superseded": frozenset({"new", "viewed", "in_progress"}), } def transition_recommendation( row: IntelligenceRecommendation, *, status: str, ignored_reason: str | None, actor_id: UUID, confirmed_savings: Decimal | None = None, realized_savings: Decimal | None = None, currency: str | None = None, ) -> None: # ``_MANUAL_STATUSES``, not ``_STATUSES``: ``superseded`` is an archival # state the re-materializer writes, never a destination a caller may name. if status not in _MANUAL_STATUSES: raise AppException(code="INTELLIGENCE_RECOMMENDATION_STATUS", message="Invalid recommendation status.", status_code=422) if status == "ignored" and not (ignored_reason or "").strip(): raise AppException(code="INTELLIGENCE_RECOMMENDATION_REASON", message="Ignoring a recommendation requires a reason.", status_code=422) current = row.status if row.status in _STATUSES else "new" if status != current and status not in _ALLOWED_TRANSITIONS[current]: raise AppException( code="INTELLIGENCE_RECOMMENDATION_TRANSITION", message=( f"A {current} recommendation cannot move to {status}." ), status_code=422, ) now = datetime.now(UTC) row.status = status row.last_changed_by_user_id = actor_id if status == "viewed" and row.viewed_at is None: row.viewed_at = now if status == "resolved" and row.resolved_at is None: # First resolution is the audited outcome date. A later savings # correction re-PATCHes ``resolved`` and must not silently move the # date it was actually resolved on. row.resolved_at = now if status == "ignored": row.ignored_at = now row.ignored_reason = (ignored_reason or "").strip()[:1000] elif current == "ignored": # Leaving the dismissal: the stale "why we dropped it" note must not # keep hanging off a row that is active again. row.ignored_reason = None if confirmed_savings is not None: row.confirmed_savings = confirmed_savings if realized_savings is not None: row.realized_savings = realized_savings if currency is not None: row.currency = currency.strip().upper()[:8] or None def workbench_path_for_dimension(dimension: str) -> str: """Where a reader goes to act on one dimension. Shared with the data-gap contract so "where do I fix this?" has ONE answer whichever way the reader arrived — a recommendation handoff or a coverage shortfall. Every value is route-verified by ``test_recommendation_handoff.py``. """ return _WORKBENCH_BY_DIMENSION.get(dimension, _CROSS_DOMAIN_WORKBENCH) __all__ = [ "MaterializationResult", "build_recommendation_handoff", "upgrade_recommendation_handoff", "workbench_path_for_dimension", "fact_dimensions", "recommendation_verification_state", "materialize_analysis_recommendations", "read_recommendation_roi", "read_recommendations", "send_recommendation_notifications", "transition_recommendation", ]