"""Zero-model read composition for Intelligence HTTP and MCP twins.""" from __future__ import annotations from collections import defaultdict from decimal import Decimal from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException from app.intelligence.access import ( IntelligenceAccessPlan, project_snapshot, ) from app.intelligence.catalog import dimension_metric_keys, public_metric_catalog from app.intelligence.models import IntelligenceAnalysis from app.intelligence.provider_registry import PLATFORM_DIMENSIONS from app.intelligence.recommendations import ( fact_dimensions, workbench_path_for_dimension, ) from app.intelligence.scoring import facts_hash, score_health from app.intelligence.snapshot_service import ( SnapshotSubject, latest_snapshot, snapshot_history, ) def _fact_dimensions(item: dict[str, Any]) -> set[str]: """Which dimensions one narrative item is grounded in. Delegates to the shared citation parser rather than re-splitting on "/": this is a VISIBILITY filter, so a spelling it fails to recognise silently hides the item from every role scoped to fewer than all eight dimensions. """ return fact_dimensions(item.get("fact_refs", [])) def _capabilities(plan: IntelligenceAccessPlan) -> dict[str, bool]: return { "can_manage_recommendations": plan.can_manage_recommendations, "can_force_generate": plan.can_force_generate, "paid_cost_authorized": plan.can_view_paid_cost, # A true authorization bit never implies that the separate paid-cost # audience exists. This remains false until that data layer ships. "paid_cost_available": False, } # A shortfall the reader can act on, rather than one word. # # "数据不足" on its own is the least useful thing this product can say: it # names no source, no coverage, no as-of time and no place to go, so the # reader's only move is to ask someone. Every gap below therefore carries # WHY it is short, WHEN its evidence was current, HOW MUCH of it exists, and # WHERE it is fixed. All four are derived deterministically from the stored # snapshot — no model is involved in producing or explaining a gap. _GAP_NO_SNAPSHOT = "no_snapshot" _GAP_PROVIDER_UNAVAILABLE = "provider_unavailable" _GAP_LOW_CONFIDENCE = "low_confidence" #: Confidence below this is reported as a gap. Mirrors the platform's own #: ``insufficient_data`` threshold in ``scoring.score_health`` so the summary #: line and the per-dimension list can never disagree. _LOW_CONFIDENCE_BELOW = Decimal("70") def _dimension_gap( dimension: str, row: dict[str, Any] | None, facts: dict[str, Any], data_as_of: str | None, ) -> dict[str, Any] | None: """Explain one dimension's shortfall, or return None when it is fine.""" if row is None: return { "dimension": dimension, "reason_code": _GAP_NO_SNAPSHOT, "data_as_of": None, "observed_metric_count": 0, "expected_metric_count": len(dimension_metric_keys(dimension)), "confidence": "0", "workbench_path": workbench_path_for_dimension(dimension), } dimension_facts = facts.get(dimension) unavailable_facts = ( dimension_facts if isinstance(dimension_facts, dict) and dimension_facts.get("available") is False else None ) metrics = row.get("metrics") or [] confidence = Decimal(str(row.get("confidence") or 0)) if unavailable_facts is None and confidence >= _LOW_CONFIDENCE_BELOW: return None reason = ( str(unavailable_facts.get("reason_code") or _GAP_PROVIDER_UNAVAILABLE) if unavailable_facts is not None else _GAP_LOW_CONFIDENCE ) return { "dimension": dimension, "reason_code": reason, # Per-dimension, not the snapshot-wide cutoff: the whole point is to # let the reader see WHICH source is stale. "data_as_of": _dimension_data_as_of(metrics) or data_as_of, "observed_metric_count": sum( 1 for metric in metrics if metric.get("score") is not None ), "expected_metric_count": len(dimension_metric_keys(dimension)), "confidence": str(confidence), "workbench_path": workbench_path_for_dimension(dimension), } def _dimension_data_as_of(metrics: list[dict[str, Any]]) -> str | None: """The stalest metric decides how current a dimension really is.""" stamps = [ str(metric["data_as_of"]) for metric in metrics if metric.get("data_as_of") ] return min(stamps) if stamps else None def _group_dimensions( plan: IntelligenceAccessPlan, ) -> dict[SnapshotSubject, list[str]]: grouped: dict[SnapshotSubject, list[str]] = defaultdict(list) for access in plan.dimensions: grouped[access.subject].append(access.dimension) return dict(grouped) async def read_overview( db: AsyncSession, *, plan: IntelligenceAccessPlan, ) -> dict[str, Any]: """Compose the latest allowed dimensions without leaking audience keys.""" requested = tuple(row.dimension for row in plan.dimensions) dimension_rows: list[dict[str, Any]] = [] aggregate_facts: dict[str, Any] = {} missing: list[str] = [] sources: list[dict[str, Any]] = [] data_times = [] created_times = [] for subject, dimensions in _group_dimensions(plan).items(): snapshot = await latest_snapshot(db, subject=subject) if snapshot is None: missing.extend(dimensions) continue projection = project_snapshot(snapshot, dimensions=dimensions).public dimension_rows.extend(projection["dimensions"]) aggregate_facts.update(projection["aggregate_facts"]) sources.append({ "snapshot_id": projection["snapshot_id"], "score_version": projection["score_version"], "facts_hash": projection["facts_hash"], "data_as_of": projection["data_as_of"], "created_at": projection["created_at"], "dimensions": dimensions, }) if snapshot.data_as_of is not None: data_times.append(snapshot.data_as_of) created_times.append(snapshot.created_at) health = score_health(dimension_rows, expected_dimensions=requested) missing_set = set(missing) scored_by_key = {row["key"]: row for row in health["dimensions"]} snapshot_as_of = min(data_times).isoformat() if data_times else None data_gaps = [ gap for gap in ( _dimension_gap( dimension, None if dimension in missing_set else scored_by_key.get(dimension), aggregate_facts, snapshot_as_of, ) for dimension in requested ) if gap is not None ] source_versions = sorted({row["score_version"] for row in sources}) version_mismatch = len(source_versions) > 1 status = ( "no_data" if not sources else ("partial" if missing or version_mismatch else "current") ) return { "status": status, "score_version": source_versions[0] if len(source_versions) == 1 else None, "score_version_mismatch": version_mismatch, "facts_hash": facts_hash({ "sources": sorted( (row["snapshot_id"], row["facts_hash"]) for row in sources ), "missing_dimensions": sorted(missing), }), "overall_score": str(health["overall_score"]), "confidence_score": str(health["confidence_score"]), "insufficient_data": ( health["insufficient_data"] or bool(missing) or version_mismatch ), "grade": health["grade"], "dimensions": health["dimensions"], "aggregate_facts": aggregate_facts, "missing_dimensions": [key for key in requested if key in missing_set], # Every shortfall, each naming its own reason, its own as-of time, its # own observed-of-expected coverage, and the surface that fixes it. "data_gaps": data_gaps, "data_as_of": snapshot_as_of, "updated_at": max(created_times).isoformat() if created_times else None, "sources": sources, "capabilities": _capabilities(plan), "model_calls": 0, } async def read_domain( db: AsyncSession, *, plan: IntelligenceAccessPlan, domain: str, ) -> dict[str, Any]: """Read one exact visible dimension, returning an explicit no-data state.""" access = next((row for row in plan.dimensions if row.dimension == domain), None) if access is None: raise AppException( code="INTELLIGENCE_DOMAIN_NOT_FOUND", message="Intelligence domain is not available.", status_code=404, ) snapshot = await latest_snapshot(db, subject=access.subject) return { "domain": domain, "status": "current" if snapshot is not None else "no_data", "metric_catalog": public_metric_catalog(domain), "snapshot": ( project_snapshot(snapshot, dimensions=[domain]).public if snapshot is not None else None ), "capabilities": _capabilities(plan), "model_calls": 0, } async def read_history( db: AsyncSession, *, plan: IntelligenceAccessPlan, limit: int, ) -> dict[str, Any]: """Return honest per-dimension series; never fabricate aligned timestamps.""" series: list[dict[str, Any]] = [] history_cache: dict[SnapshotSubject, list[Any]] = {} for access in plan.dimensions: rows = history_cache.get(access.subject) if rows is None: rows = await snapshot_history(db, subject=access.subject, limit=limit) history_cache[access.subject] = rows points = [] for snapshot in rows: projected = project_snapshot(snapshot, dimensions=[access.dimension]).public points.append({ "snapshot_id": projected["snapshot_id"], "score_version": projected["score_version"], "facts_hash": projected["facts_hash"], "score": projected["overall_score"], "confidence_score": projected["confidence_score"], "insufficient_data": projected["insufficient_data"], "grade": projected["grade"], "data_as_of": projected["data_as_of"], "created_at": projected["created_at"], }) series.append({"dimension": access.dimension, "points": points}) return { "series": series, "limit": max(1, min(365, limit)), "capabilities": _capabilities(plan), "model_calls": 0, } async def read_latest_analysis( db: AsyncSession, *, plan: IntelligenceAccessPlan, ) -> dict[str, Any]: """Read the latest shared platform brief without invoking the model.""" platform_subject = next( (item.subject for item in plan.dimensions if item.subject.subject_type == "platform"), None, ) if platform_subject is None: return {"status": "no_data", "analysis": None, "model_calls": 0} row = (await db.execute( select(IntelligenceAnalysis) .where( IntelligenceAnalysis.subject_type == platform_subject.subject_type, IntelligenceAnalysis.subject_id == platform_subject.subject_id, IntelligenceAnalysis.audience_key == platform_subject.audience_key, IntelligenceAnalysis.analysis_kind == "platform_daily_brief", IntelligenceAnalysis.status == "succeeded", ) .order_by(IntelligenceAnalysis.generated_at.desc()) .limit(1) )).scalar_one_or_none() if row is None: return {"status": "no_data", "analysis": None, "model_calls": 0} allowed = {item.dimension for item in plan.dimensions} all_dimensions = allowed == set(PLATFORM_DIMENSIONS) def visible_items(items: list[dict[str, Any]] | None) -> list[dict[str, Any]]: if all_dimensions: return items or [] return [item for item in (items or []) if _fact_dimensions(item) & allowed] analysis: dict[str, Any] = { "id": str(row.id), "status": row.status, "summary_zh": row.summary_zh if all_dimensions else None, "summary_en": row.summary_en if all_dimensions else None, "conclusions": visible_items(row.conclusions), "risks": visible_items(row.risks), "opportunities": visible_items(row.opportunities), "forecasts": visible_items(row.forecasts), "data_gaps": visible_items(row.data_gaps), "scope_limited": not all_dimensions, "facts_hash": row.facts_hash, "data_as_of": row.data_as_of.isoformat() if row.data_as_of else None, "generated_at": row.generated_at.isoformat() if row.generated_at else None, "cache_reused": row.cache_reused, } # Model identity and token counts are the same triple the cloud-scope twin # (``cloud_scope_projection._generation_meta``) withholds from every # non-privileged reader — and the platform brief admits a much wider # audience than that twin does: HR, finance, auditor, cloud_admin and both # asset-admin tiers all reach this read for their own dimensions. They are # admitted to the brief's CONTENT, not to what the platform spends running # it. Two surfaces over the same table cannot disagree about whether that # is public, so the stricter posture wins and the operator tier # (``can_manage_recommendations`` — global_admin) is the one that sees it. if plan.can_manage_recommendations: analysis["model_used"] = row.model_used analysis["input_token_count"] = row.input_token_count analysis["output_token_count"] = row.output_token_count return { "status": "current", "analysis": analysis, "model_calls": 0, } __all__ = ["read_domain", "read_history", "read_latest_analysis", "read_overview"]