"""Public, role-trimmed projection for shared cloud-scope analysis. Model output contains anonymous references only. This module is the sole place that may resolve those references back to local objects, and it does so against the already-authorized scope plus the caller's exact access shape. """ from __future__ import annotations import uuid from collections.abc import Iterable from datetime import datetime from decimal import Decimal from typing import Any from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.models import ( SCOPE_STATUS_CLOSED, SCOPE_TYPE_AZURE_RESOURCE_GROUP, CmdbCloudAccessGrant, CmdbCloudResource, CmdbCostFact, CmdbServer, ) from app.domains.cmdb.scope_billing_detail import scope_fact_filters from app.identity.person_models import IdentityPerson from app.intelligence.cloud_scope_analysis import ( CLOUD_SCOPE_ANALYSIS_KIND, CloudScopeAnalysisAccess, CloudScopeAnalysisState, ) from app.intelligence.cloud_scope_scoring import public_cloud_scope_metric_catalog from app.intelligence.models import IntelligenceAnalysis, IntelligenceSnapshot from app.intelligence.provider_registry import BUSINESS_AUDIENCE_KEY from app.intelligence.scoring import json_value _NARRATIVE_COLLECTIONS = ( "conclusions", "risks", "opportunities", "forecasts", "recommendations", "data_gaps", ) def _iso(value: datetime | None) -> str | None: return value.isoformat() if value is not None else None def _score(value: Decimal | int | float) -> str: return str(value) def project_cloud_scope_snapshot(snapshot: IntelligenceSnapshot) -> dict[str, Any]: """Expose deterministic health without the server-only reference map.""" if snapshot.subject_type != "cloud_scope" or snapshot.audience_key != BUSINESS_AUDIENCE_KEY: raise ValueError("Not a shared cloud-scope snapshot") return { "snapshot_id": str(snapshot.id), "score_version": snapshot.score_version, "facts_hash": snapshot.facts_hash, "overall_score": _score(snapshot.overall_score), "confidence_score": _score(snapshot.confidence_score), "insufficient_data": Decimal(str(snapshot.confidence_score)) < 70, "grade": snapshot.grade, "dimensions": json_value(snapshot.dimensions, decimal_as_string=True), "metric_catalog": public_cloud_scope_metric_catalog(), "aggregate_facts": json_value( snapshot.aggregate_facts, decimal_as_string=True, ), "data_as_of": _iso(snapshot.data_as_of), "created_at": _iso(snapshot.created_at), } def _snapshot_belongs_to_access( snapshot: IntelligenceSnapshot, access: CloudScopeAnalysisAccess, ) -> bool: return bool( snapshot.subject_type == "cloud_scope" and snapshot.subject_id == str(access.scope.id) and snapshot.audience_key == BUSINESS_AUDIENCE_KEY ) def _analysis_belongs_to_access( analysis: IntelligenceAnalysis, access: CloudScopeAnalysisAccess, ) -> bool: return bool( analysis.subject_type == "cloud_scope" and analysis.subject_id == str(access.scope.id) and analysis.audience_key == BUSINESS_AUDIENCE_KEY and analysis.analysis_kind == CLOUD_SCOPE_ANALYSIS_KIND ) def _used_evidence_refs(analysis: IntelligenceAnalysis) -> set[str]: used: set[str] = set() for collection in _NARRATIVE_COLLECTIONS: rows = getattr(analysis, collection, None) or [] for row in rows: if not isinstance(row, dict): continue refs = row.get("evidence_refs") if isinstance(refs, list): used.update(value for value in refs if isinstance(value, str)) return used def _metric_evidence_refs(snapshot: IntelligenceSnapshot) -> set[str]: """Return only the anonymous references used by deterministic metrics.""" used: set[str] = set() for dimension in snapshot.dimensions or []: if not isinstance(dimension, dict): continue for metric in dimension.get("metrics") or []: if not isinstance(metric, dict): continue refs = metric.get("evidence_refs") if isinstance(refs, list): used.update(value for value in refs if isinstance(value, str)) return used async def _people_labels( db: AsyncSession, person_ids: Iterable[str | uuid.UUID | None], ) -> dict[str, str]: parsed: list[uuid.UUID] = [] for value in person_ids: if value is None: continue try: parsed.append(uuid.UUID(str(value))) except ValueError: continue if not parsed: return {} rows = ( await db.execute( select(IdentityPerson.id, IdentityPerson.display_name).where( IdentityPerson.id.in_(parsed), IdentityPerson.deleted_at.is_(None), ) ) ).all() return {str(row.id): row.display_name or "" for row in rows} async def _project_evidence_refs( db: AsyncSession, *, snapshot: IntelligenceSnapshot, evidence_refs: set[str], access: CloudScopeAnalysisAccess, person_id: uuid.UUID, ) -> list[dict[str, Any]]: local = snapshot.reference_map or {} requested = {ref: local.get(ref) for ref in sorted(evidence_refs)} grant_ids: list[uuid.UUID] = [] server_ids: list[uuid.UUID] = [] directory_resource_ids: list[uuid.UUID] = [] for mapping in requested.values(): if not isinstance(mapping, dict): continue kind = mapping.get("kind") if kind not in {"grant", "server", "resource"}: continue try: parsed = uuid.UUID(str(mapping.get("object_id"))) except (TypeError, ValueError): continue if kind == "grant": grant_ids.append(parsed) elif kind == "server": server_ids.append(parsed) else: directory_resource_ids.append(parsed) grants = ( list( ( await db.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.id.in_(grant_ids), CmdbCloudAccessGrant.scope_id == str(access.scope.id), ) ) ).scalars() ) if grant_ids else [] ) grant_map = {str(row.id): row for row in grants} servers = ( list( ( await db.execute( select(CmdbServer).where( CmdbServer.id.in_(server_ids), CmdbServer.integration_id == str(access.scope.integration_id), CmdbServer.deleted_at.is_(None), ) ) ).scalars() ) if server_ids else [] ) server_map = {str(row.id): row for row in servers} directory_scope_clause = ( or_( CmdbCloudResource.scope_external_id == access.scope.external_id, CmdbCloudResource.raw_metadata["resource_group"].as_string() == access.scope.external_id, ) if access.scope.scope_type == SCOPE_TYPE_AZURE_RESOURCE_GROUP else CmdbCloudResource.scope_external_id == access.scope.external_id ) directory_resources = ( list( ( await db.scalars( select(CmdbCloudResource).where( CmdbCloudResource.id.in_(directory_resource_ids), CmdbCloudResource.integration_id == str(access.scope.integration_id), directory_scope_clause, CmdbCloudResource.deleted_at.is_(None), ) ) ).all() ) if directory_resource_ids else [] ) directory_resource_map = {str(row.id): row for row in directory_resources} cost_filters, _ = scope_fact_filters(access.scope) resource_candidates = {row.instance_id for row in servers if row.instance_id} service_candidates: set[str] = set() for mapping in requested.values(): if not isinstance(mapping, dict): continue object_id = str(mapping.get("object_id") or "") expected_prefix = f"{access.scope.integration_id}:" if not object_id.startswith(expected_prefix): continue value = object_id[len(expected_prefix) :] if mapping.get("kind") == "costresource": resource_candidates.add(value) elif mapping.get("kind") == "costservice": service_candidates.add(value) attributed_resources: set[str] = set() attributed_services: set[str] = set() if cost_filters and (resource_candidates or service_candidates): service_label = func.coalesce(CmdbCostFact.service, "unclassified") candidates = [] if resource_candidates: candidates.append(CmdbCostFact.resource_id.in_(resource_candidates)) if service_candidates: candidates.append(service_label.in_(service_candidates)) cost_rows = ( await db.execute( select(CmdbCostFact.resource_id, service_label.label("service")) .where(*cost_filters, or_(*candidates)) .distinct() ) ).all() attributed_resources = { str(row.resource_id) for row in cost_rows if row.resource_id is not None } attributed_services = {str(row.service) for row in cost_rows} server_map = { key: row for key, row in server_map.items() if row.instance_id in attributed_resources } labels = await _people_labels(db, (row.person_id for row in grants)) privileged_grants = access.privileged_reader or access.direct_owner # Spelled out per branch rather than interpolated from a ``prefix`` # variable: the BE↔FE deep-link contract scanner reads path literals, and # ``f"{prefix}/servers/{id}"`` reached it as the un-routable ``/servers/:id`` # — which is how the privileged ``/admin/cloud/servers/:id`` evidence link # went unclassified, and unmapped, so tapping it in Feishu H5 did nothing. privileged_href = access.privileged_reader scope_href = ( f"/admin/cloud/scopes/{access.scope.id}" if privileged_href else f"/my/cloud/scopes/{access.scope.id}" ) def _server_href(server_id: Any) -> str: return ( f"/admin/cloud/servers/{server_id}" if privileged_href else f"/my/cloud/servers/{server_id}" ) projected: list[dict[str, Any]] = [] for ref, mapping in requested.items(): hidden = {"ref": ref, "kind": "unknown", "visible": False} if not isinstance(mapping, dict): projected.append(hidden) continue kind = str(mapping.get("kind") or "unknown") object_id = str(mapping.get("object_id") or "") if kind == "scope" and object_id == str(access.scope.id): projected.append( { "ref": ref, "kind": kind, "visible": True, "object_id": object_id, "display_label": access.scope.display_name or "", "href": scope_href, } ) elif kind == "grant": grant = grant_map.get(object_id) may_view = grant is not None and ( privileged_grants or str(grant.person_id or "") == str(person_id) ) if not may_view or grant is None: projected.append({"ref": ref, "kind": kind, "visible": False}) continue role = grant.role_display_name or grant.role_key person = labels.get(str(grant.person_id or ""), "") display = f"{person} · {role}" if person and privileged_grants else role projected.append( { "ref": ref, "kind": kind, "visible": True, "object_id": object_id, "display_label": display, "href": scope_href, } ) elif kind == "server": server = server_map.get(object_id) if server is None: projected.append({"ref": ref, "kind": kind, "visible": False}) continue projected.append( { "ref": ref, "kind": kind, "visible": True, "object_id": object_id, "display_label": server.name or server.instance_id, "href": _server_href(server.id), } ) elif kind == "resource": resource = directory_resource_map.get(object_id) if resource is None: projected.append({"ref": ref, "kind": kind, "visible": False}) continue projected.append( { "ref": ref, "kind": kind, "visible": True, "object_id": object_id, "display_label": resource.name or resource.provider_resource_id, "href": f"{scope_href}?tab=resources", } ) elif kind in {"costservice", "costresource"}: expected_prefix = f"{access.scope.integration_id}:" if not object_id.startswith(expected_prefix): projected.append({"ref": ref, "kind": kind, "visible": False}) continue label = object_id[len(expected_prefix) :] attributed = ( label in attributed_services if kind == "costservice" else label in attributed_resources ) if not attributed: projected.append({"ref": ref, "kind": kind, "visible": False}) continue projected.append( { "ref": ref, "kind": kind, "visible": True, # Integration identity remains server-only; the suffix is the # exact service/resource the authorized billing drawer shows. "object_id": label, "display_label": label, "href": scope_href, } ) else: projected.append({"ref": ref, "kind": kind, "visible": False}) return projected def _generation_meta( row: IntelligenceAnalysis, *, generator_label: str | None, privileged: bool, request_cache_reused: bool | None = None, ) -> dict[str, Any]: meta: dict[str, Any] = { "analysis_id": str(row.id), "status": row.status, "facts_hash": row.facts_hash, "data_as_of": _iso(row.data_as_of), "generated_at": _iso(row.generated_at), "created_at": _iso(row.created_at), "forced": bool(row.forced), "cache_reused": ( bool(request_cache_reused) if request_cache_reused is not None else bool(row.cache_reused) ), "failure_code": row.failure_code if row.status == "failed" else None, } if privileged: meta.update( { # A real colleague's display name. It sat in the base dict # above, one line under the comment explaining why the model # and token counts are withheld — so every grantee of a # subscription learned which named person last ran the # analysis (and, through ``latest_attempt``, who last failed # to). That is directory data about a third party attached to # an operational event, and it belongs in the same privileged # block as the rest of the generation metadata. "generated_by": generator_label, "model_used": row.model_used, "input_token_count": row.input_token_count, "output_token_count": row.output_token_count, # ``failure_message`` is deliberately NOT projected, not even # for privileged readers. It is the one stored field that can # quote the model or the provider verbatim, and #2799 named # that intent in its own fixture ("private provider diagnostic # must never escape"). Running it back through # ``assert_llm_safe`` was considered and rejected: that guard # is written for the OUTBOUND direction and has known coverage # holes (no IPv6, a fixed TLD list, bare hostnames), so it # cannot be the thing standing between a vendor sentence and a # reader. The operator need is real — telling an invented # number from a schema drift — but it should be met with an # enumerated reason derived server-side, not by forwarding # free text. ``failure_code`` remains visible to everyone. } ) return meta async def _analysis_snapshot( db: AsyncSession, analysis: IntelligenceAnalysis, ) -> IntelligenceSnapshot | None: snapshot = await db.get(IntelligenceSnapshot, analysis.snapshot_id) if snapshot is None: return None if ( snapshot.subject_type != "cloud_scope" or snapshot.subject_id != analysis.subject_id or snapshot.audience_key != BUSINESS_AUDIENCE_KEY or snapshot.facts_hash != analysis.facts_hash ): return None return snapshot async def project_cloud_scope_analysis_state( db: AsyncSession, *, state: CloudScopeAnalysisState, person_id: uuid.UUID, request_cache_reused: bool | None = None, ) -> dict[str, Any]: """Compose current health, last valid analysis and last-attempt state.""" successful = ( state.latest_successful if state.latest_successful is not None and _analysis_belongs_to_access(state.latest_successful, state.access) else None ) attempt = ( state.latest_attempt if state.latest_attempt is not None and _analysis_belongs_to_access(state.latest_attempt, state.access) else None ) current_snapshot = ( state.snapshot if state.snapshot is not None and _snapshot_belongs_to_access(state.snapshot, state.access) else None ) generator_ids = [ row.generated_by_user_id for row in (successful, attempt) if row is not None and row.generated_by_user_id is not None ] labels = await _people_labels(db, generator_ids) analysis_payload = None integrity_status = ( "current_snapshot_invalid" if state.snapshot is not None and current_snapshot is None else "valid" ) if state.latest_successful is not None and successful is None: integrity_status = "analysis_scope_invalid" elif state.latest_attempt is not None and attempt is None: integrity_status = "latest_attempt_scope_invalid" if successful is not None: source_snapshot = await _analysis_snapshot(db, successful) if source_snapshot is None: integrity_status = "analysis_snapshot_invalid" else: analysis_payload = { "analysis_id": str(successful.id), "summary_zh": successful.summary_zh, "summary_en": successful.summary_en, "summary_fact_refs": successful.summary_fact_refs or [], **{ key: json_value(getattr(successful, key) or []) for key in _NARRATIVE_COLLECTIONS }, "generation": _generation_meta( successful, generator_label=labels.get(str(successful.generated_by_user_id)), privileged=state.access.privileged_reader, request_cache_reused=request_cache_reused, ), "evidence": await _project_evidence_refs( db, snapshot=source_snapshot, evidence_refs=_used_evidence_refs(successful), access=state.access, person_id=person_id, ), "facts_current": bool( current_snapshot and successful.facts_hash == current_snapshot.facts_hash ), } latest_attempt = ( _generation_meta( attempt, generator_label=labels.get(str(attempt.generated_by_user_id)), privileged=state.access.privileged_reader, request_cache_reused=( request_cache_reused if successful is not None and attempt.id == successful.id else None ), ) if attempt is not None else None ) if request_cache_reused and analysis_payload is not None: status = "ready" elif attempt is not None and attempt.status == "pending": status = "generating" elif attempt is not None and attempt.status == "failed": status = "refresh_failed" if analysis_payload else "analysis_failed" elif analysis_payload is not None: status = "ready" elif current_snapshot is not None: status = "scored" else: status = "never_analyzed" health_payload = ( project_cloud_scope_snapshot(current_snapshot) if current_snapshot is not None else None ) if health_payload is not None and current_snapshot is not None: health_payload["evidence"] = await _project_evidence_refs( db, snapshot=current_snapshot, evidence_refs=_metric_evidence_refs(current_snapshot), access=state.access, person_id=person_id, ) # Which of the two gates in ``generate_cloud_scope_analysis`` is shut, in # the order that function raises them: the spend check runs first, so a # broad reader looking at a CLOSED subscription is refused for the reason # the server would actually give. ``None`` means the button is live. # # Without this the client only received ``can_generate: false`` and could # not tell "you may read this analysis but not spend tokens on this # subscription" from "this subscription is closed" — so it said nothing at # all next to a greyed-out button, which reads as a broken page. if not state.access.can_spend_model_tokens: generate_blocked_reason: str | None = "spend_forbidden" elif state.access.scope.status == SCOPE_STATUS_CLOSED: generate_blocked_reason = "scope_closed" else: generate_blocked_reason = None return { "status": status, "integrity_status": integrity_status, "scope_id": str(state.access.scope.id), "health": health_payload, "analysis": analysis_payload, "latest_attempt": latest_attempt, "capabilities": { # Truthful about spend authority, not just about the scope being # open: a broad reader (auditor / finance / management) with no # relationship to this subscription is refused by # ``generate_cloud_scope_analysis``, so the button must be off. "can_generate": generate_blocked_reason is None, "can_force_generate": state.access.can_force, "generate_blocked_reason": generate_blocked_reason, "shared_result": True, }, } async def project_cloud_scope_analysis_history( db: AsyncSession, *, access: CloudScopeAnalysisAccess, rows: list[IntelligenceAnalysis], ) -> dict[str, Any]: """Project bounded history summaries without multiplying full narratives.""" snapshot_ids = {row.snapshot_id for row in rows} snapshots = ( list( ( await db.execute( select(IntelligenceSnapshot).where( IntelligenceSnapshot.id.in_(snapshot_ids), ) ) ).scalars() ) if snapshot_ids else [] ) snapshot_map = {row.id: row for row in snapshots} labels = await _people_labels( db, (row.generated_by_user_id for row in rows), ) items: list[dict[str, Any]] = [] for row in rows: if not _analysis_belongs_to_access(row, access): items.append( { "status": "invalid", "integrity_status": "analysis_scope_invalid", } ) continue snapshot = snapshot_map.get(row.snapshot_id) valid = bool( snapshot and _snapshot_belongs_to_access(snapshot, access) and snapshot.facts_hash == row.facts_hash ) items.append( { "analysis_id": str(row.id), "status": row.status, "summary_zh": row.summary_zh if valid and row.status == "succeeded" else None, "summary_en": row.summary_en if valid and row.status == "succeeded" else None, "overall_score": _score(snapshot.overall_score) if valid and snapshot else None, "confidence_score": ( _score(snapshot.confidence_score) if valid and snapshot else None ), "grade": snapshot.grade if valid and snapshot else None, "integrity_status": "valid" if valid else "snapshot_invalid", "generation": _generation_meta( row, generator_label=labels.get(str(row.generated_by_user_id)), privileged=access.privileged_reader, ), } ) return {"items": items, "count": len(items), "shared_result": True} __all__ = [ "project_cloud_scope_analysis_history", "project_cloud_scope_analysis_state", "project_cloud_scope_snapshot", ]