import asyncio import json import uuid from datetime import UTC, datetime, timedelta from decimal import Decimal import pytest from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.core.exceptions import AppException from app.integrations.models import PlatformSetting from app.intelligence.contracts import ( IntelligenceContractError, validate_intelligence_output, ) from app.intelligence.generation import ( _consume_daily_quota, _release_lease, _reusable_analysis, _try_acquire_lease, build_model_fact_package, generate_analysis, ) from app.intelligence.models import IntelligenceAnalysis, IntelligenceSnapshot from app.intelligence.settings import SETTING_DAILY_LIMIT def _item(**overrides): value = { "title_zh": "预算使用率 80%", "title_en": "Budget utilization is 80%", "detail_zh": "本月成本为 1200 元。", "detail_en": "Current-month cost is 1,200.", "fact_refs": ["/facts/budget_utilization", "/facts/month_cost"], "evidence_refs": ["scope_01"], "confidence": "high", } value.update(overrides) return value def _valid_output(): return { "summary_zh": "本月成本 1200 元，预算使用率 80%。", "summary_en": "Current-month cost is 1,200 and budget utilization is 80%.", "summary_fact_refs": ["/facts/month_cost", "/facts/budget_utilization"], "conclusions": [_item()], "risks": [{**_item(), "severity": "medium"}], "opportunities": [], "forecasts": [], "recommendations": [{**_item(), "priority": "p1", "action_kind": "review"}], "data_gaps": [], "_meta": {"input_tokens": 100, "output_tokens": 200}, } @pytest.fixture def fact_package(): return { "overall_score": 80, "confidence_score": 95, "facts": {"month_cost": 1200, "budget_utilization": 80}, "evidence_refs": ["scope_01"], "horizon_days": [30, 60, 90], } def test_valid_contract_is_bilingual_grounded_and_strips_transport_meta(fact_package): result = validate_intelligence_output( _valid_output(), facts=fact_package, allowed_evidence_refs={"scope_01"}, confidence_score=95, ) assert result.summary_zh.startswith("本月成本") assert result.summary_en.startswith("Current-month") assert not hasattr(result, "_meta") def test_contract_rejects_invented_number_even_when_json_is_valid(fact_package): raw = _valid_output() raw["summary_zh"] = "本月成本 9999 元。" with pytest.raises(IntelligenceContractError, match="uncited numbers"): validate_intelligence_output( raw, facts=fact_package, allowed_evidence_refs={"scope_01"}, confidence_score=95, ) def test_contract_scans_numbers_touching_chinese_characters(fact_package): raw = _valid_output() raw["summary_zh"] = "本月成本9999元。" with pytest.raises(IntelligenceContractError, match="9999"): validate_intelligence_output( raw, facts=fact_package, allowed_evidence_refs={"scope_01"}, confidence_score=95, ) def test_contract_rejects_fact_pointer_to_container(fact_package): raw = _valid_output() raw["summary_fact_refs"] = ["/facts"] with pytest.raises(IntelligenceContractError, match="scalar"): validate_intelligence_output( raw, facts=fact_package, allowed_evidence_refs={"scope_01"}, confidence_score=95, ) def test_contract_rejects_unknown_object_reference(fact_package): raw = _valid_output() raw["risks"][0]["evidence_refs"] = ["server_99"] with pytest.raises(IntelligenceContractError, match="unknown evidence refs"): validate_intelligence_output( raw, facts=fact_package, allowed_evidence_refs={"scope_01"}, confidence_score=95, ) def test_contract_rejects_extra_untyped_output(fact_package): raw = _valid_output() raw["secret_advice"] = "uncontracted" with pytest.raises(IntelligenceContractError, match="Invalid intelligence JSON"): validate_intelligence_output( raw, facts=fact_package, allowed_evidence_refs={"scope_01"}, confidence_score=95, ) def test_low_confidence_requires_data_gap_and_neutralises_strong_claims( fact_package, ): """The invariant is unchanged; the mechanism deliberately is not. A high-confidence claim on low-confidence facts used to REFUSE the whole document. Production (2026-08-17) threw away an otherwise grounded cloud subscription narrative over exactly that, so the claim is now clamped down to ``medium`` instead — which can only ever LOWER a claim, so "low-confidence facts must not support high-confidence conclusions" still holds, and now holds deterministically rather than depending on the model obeying an instruction it was already given. The data-gaps half stays a refusal: a missing explanation cannot be manufactured, only the model can supply one. """ with pytest.raises(IntelligenceContractError, match="data gap"): validate_intelligence_output( _valid_output(), facts=fact_package, allowed_evidence_refs={"scope_01"}, confidence_score=60, ) raw = _valid_output() raw["data_gaps"] = [_item(confidence="low")] result = validate_intelligence_output( raw, facts=fact_package, allowed_evidence_refs={"scope_01"}, confidence_score=60, ) assert not [ item for collection in ( result.conclusions, result.risks, result.opportunities, result.forecasts, result.recommendations, result.data_gaps, ) for item in collection if item.confidence == "high" ], "no high-confidence claim may survive on low-confidence facts" assert result.data_gaps[0].confidence == "low" # And a genuinely high-confidence package keeps its high claims. confident = validate_intelligence_output( _valid_output(), facts=fact_package, allowed_evidence_refs={"scope_01"}, confidence_score=95, ) assert any( item.confidence == "high" for item in confident.conclusions ), "clamping must not fire when the facts support the claim" def test_model_package_excludes_subject_and_local_reference_mapping(): snapshot = IntelligenceSnapshot( subject_type="cloud_scope", subject_id="real-scope-uuid", audience_key="shared", score_version="health-v1", facts_hash="a" * 64, overall_score=Decimal("81.25"), confidence_score=Decimal("92.50"), grade="healthy", dimensions=[{"key": "cloud_network", "score": 80}], aggregate_facts={ "resource_count": 12, "idle_rate": 5, "precise_cost": Decimal("1234567890123456.1234"), }, reference_map={ "srv_01": {"kind": "server", "object_id": "private-object-uuid"}, }, data_as_of=datetime(2026, 8, 15, tzinfo=UTC), ) package = build_model_fact_package(snapshot) serialized = json.dumps(package) assert package["evidence_refs"] == ["srv_01"] assert "subject_id" not in package assert "real-scope-uuid" not in serialized assert "private-object-uuid" not in serialized assert package["overall_score"] == "81.25" assert package["facts"]["precise_cost"] == "1234567890123456.1234" def _snapshot(subject_type: str, subject_id: str, fact_hash: str) -> IntelligenceSnapshot: return IntelligenceSnapshot( subject_type=subject_type, subject_id=subject_id, audience_key="shared", score_version="health-v1", facts_hash=fact_hash, overall_score=80, confidence_score=90, grade="healthy", dimensions=[], aggregate_facts={"count": 1}, reference_map={}, data_as_of=datetime.now(UTC), ) def _analysis(snapshot: IntelligenceSnapshot, generated_at: datetime) -> IntelligenceAnalysis: return IntelligenceAnalysis( snapshot_id=snapshot.id, subject_type=snapshot.subject_type, subject_id=snapshot.subject_id, audience_key=snapshot.audience_key, analysis_kind="brief", facts_hash=snapshot.facts_hash, generation_key=uuid.uuid4().hex, status="succeeded", summary_zh="结论", summary_en="Conclusion", data_as_of=snapshot.data_as_of, generated_at=generated_at, ) @pytest.mark.asyncio async def test_platform_cache_reuses_unchanged_facts_without_repeating_daily(db_session): old = _snapshot("platform", "platform", "a" * 64) db_session.add(old) await db_session.flush() analysis = _analysis(old, datetime.now(UTC) - timedelta(days=60)) db_session.add(analysis) await db_session.commit() reused = await _reusable_analysis( db_session, old, "brief", cloud_cache_days=7, ) assert reused is not None assert reused.id == analysis.id @pytest.mark.asyncio async def test_all_empty_snapshot_cannot_spend_model_tokens(db_session): snapshot = _snapshot("platform", "platform", "0" * 64) snapshot.data_as_of = None db_session.add(snapshot) await db_session.commit() with pytest.raises(AppException) as error: await generate_analysis( db_session, snapshot=snapshot, analysis_kind="brief", actor_id=uuid.uuid4(), ) assert error.value.code == "INTELLIGENCE_NO_DATA" assert await db_session.scalar(select(func.count()).select_from(IntelligenceAnalysis)) == 0 @pytest.mark.asyncio async def test_platform_cache_caps_changed_facts_at_one_success_per_local_day(db_session): old = _snapshot("platform", "platform", "b" * 64) current = _snapshot("platform", "platform", "c" * 64) db_session.add_all([old, current]) await db_session.flush() today_analysis = _analysis(old, datetime.now(UTC)) db_session.add(today_analysis) await db_session.commit() reused = await _reusable_analysis( db_session, current, "brief", cloud_cache_days=7, ) assert reused is not None assert reused.id == today_analysis.id @pytest.mark.asyncio async def test_cloud_cache_requires_same_facts_within_seven_days(db_session): old = _snapshot("cloud_scope", "scope-a", "d" * 64) current = _snapshot("cloud_scope", "scope-a", "e" * 64) db_session.add_all([old, current]) await db_session.flush() db_session.add(_analysis(old, datetime.now(UTC) - timedelta(days=2))) await db_session.commit() assert await _reusable_analysis( db_session, current, "brief", cloud_cache_days=7, ) is None expired_matching = _analysis(current, datetime.now(UTC) - timedelta(days=8)) db_session.add(expired_matching) await db_session.commit() assert await _reusable_analysis( db_session, current, "brief", cloud_cache_days=7, ) is None fresh_matching = _analysis(current, datetime.now(UTC) - timedelta(days=6)) db_session.add(fresh_matching) await db_session.commit() reused = await _reusable_analysis( db_session, current, "brief", cloud_cache_days=7, ) assert reused is not None assert reused.id == fresh_matching.id @pytest.mark.asyncio async def test_generation_lease_allows_exactly_one_concurrent_owner(db_engine, monkeypatch): from app.core import database factory = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False) monkeypatch.setattr(database, "async_session_factory", factory) first, second = await asyncio.gather( _try_acquire_lease("same-subject"), _try_acquire_lease("same-subject"), ) owners = [owner for owner in (first, second) if owner is not None] assert len(owners) == 1 await _release_lease("same-subject", owners[0]) assert await _try_acquire_lease("same-subject") is not None @pytest.mark.asyncio async def test_daily_quota_is_atomic_under_concurrency(db_engine, db_session, monkeypatch): from app.core import database factory = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False) monkeypatch.setattr(database, "async_session_factory", factory) db_session.add(PlatformSetting(key=SETTING_DAILY_LIMIT, value=1)) await db_session.commit() actor_id = uuid.uuid4() results = await asyncio.gather( _consume_daily_quota(actor_id), _consume_daily_quota(actor_id), return_exceptions=True, ) failures = [result for result in results if isinstance(result, AppException)] assert len(failures) == 1 assert failures[0].code == "INTELLIGENCE_DAILY_QUOTA" @pytest.mark.asyncio async def test_generation_persists_tokens_and_reuses_cache_without_second_call( db_engine, db_session, monkeypatch, ): from app.core import database from app.intelligence import generation factory = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False) monkeypatch.setattr(database, "async_session_factory", factory) snapshot = _snapshot("cloud_scope", "scope-generation", "f" * 64) snapshot.aggregate_facts = {"month_cost": 1200, "budget_utilization": 80} snapshot.reference_map = {"scope_01": {"kind": "scope", "object_id": "local-id"}} db_session.add_all([ snapshot, PlatformSetting(key="ai_spending.llm.api_key", value="test-key"), ]) await db_session.commit() class FakeClient: calls = 0 def __init__(self, *args, **kwargs): pass async def chat_json(self, **kwargs): self.__class__.calls += 1 return { **_valid_output(), "_meta": { "model_used": "fake-grounded-model", "input_tokens": 123, "output_tokens": 45, }, } monkeypatch.setattr(generation, "BailianLLMClient", FakeClient) first = await generate_analysis( db_session, snapshot=snapshot, analysis_kind="cloud_scope", actor_id=None, enforce_user_quota=False, ) assert first.cache_reused is False assert first.analysis.status == "succeeded" assert first.analysis.input_token_count == 123 assert first.analysis.output_token_count == 45 assert first.analysis.model_used == "fake-grounded-model" assert first.analysis.facts_hash == snapshot.facts_hash assert first.analysis.summary_fact_refs == [ "/facts/month_cost", "/facts/budget_utilization", ] second = await generate_analysis( db_session, snapshot=snapshot, analysis_kind="cloud_scope", actor_id=None, enforce_user_quota=False, ) assert second.cache_reused is True assert second.analysis.id == first.analysis.id assert FakeClient.calls == 1 forced = await generate_analysis( db_session, snapshot=snapshot, analysis_kind="cloud_scope", actor_id=None, force=True, enforce_user_quota=False, ) assert forced.cache_reused is False assert forced.analysis.id != first.analysis.id assert forced.analysis.forced is True assert FakeClient.calls == 2 @pytest.mark.asyncio async def test_invalid_model_output_is_retained_as_failed_without_harming_snapshot( db_engine, db_session, monkeypatch, ): from app.core import database from app.intelligence import generation factory = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False) monkeypatch.setattr(database, "async_session_factory", factory) snapshot = _snapshot("platform", "platform", "9" * 64) snapshot.aggregate_facts = {"month_cost": 1200, "budget_utilization": 80} snapshot.reference_map = {"scope_01": {"kind": "scope", "object_id": "local-id"}} db_session.add_all([ snapshot, PlatformSetting(key="ai_spending.llm.api_key", value="test-key"), ]) await db_session.commit() snapshot_id = snapshot.id class InventingClient: def __init__(self, *args, **kwargs): pass async def chat_json(self, **kwargs): raw = _valid_output() raw["summary_zh"] = "系统已节省9999元。" return raw monkeypatch.setattr(generation, "BailianLLMClient", InventingClient) with pytest.raises(AppException) as caught: await generate_analysis( db_session, snapshot=snapshot, analysis_kind="platform_brief", actor_id=None, enforce_user_quota=False, ) assert caught.value.code == "INTELLIGENCE_OUTPUT_INVALID" failed = (await db_session.execute( select(IntelligenceAnalysis).where(IntelligenceAnalysis.snapshot_id == snapshot_id) )).scalar_one() assert failed.status == "failed" assert failed.failure_code == "INTELLIGENCE_OUTPUT_INVALID" assert failed.summary_zh is None # ``_VALIDATION_ATTEMPTS`` asks, because a contract rejection is # re-asked — this client invents on every reply, so all three are spent # and ALL THREE are audited. 100/200 here would mean the row # under-reports the spend of exactly the runs that cost the most. assert failed.input_token_count == 100 * generation._VALIDATION_ATTEMPTS assert failed.output_token_count == 200 * generation._VALIDATION_ATTEMPTS assert failed.facts_hash == "9" * 64 assert await db_session.get(IntelligenceSnapshot, snapshot_id) is not None @pytest.mark.asyncio async def test_unsafe_fact_package_never_calls_model_and_is_audited( db_engine, db_session, monkeypatch, ): from app.core import database from app.intelligence import generation factory = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False) monkeypatch.setattr(database, "async_session_factory", factory) snapshot = _snapshot("platform", "unsafe-platform", "8" * 64) snapshot.aggregate_facts = {"email": "person@example.com"} db_session.add(snapshot) await db_session.commit() snapshot_id = snapshot.id class MustNotRun: def __init__(self, *args, **kwargs): raise AssertionError("privacy rejection must happen before client creation") monkeypatch.setattr(generation, "BailianLLMClient", MustNotRun) with pytest.raises(AppException) as caught: await generate_analysis( db_session, snapshot=snapshot, analysis_kind="platform_brief", actor_id=None, enforce_user_quota=False, ) assert caught.value.code == "INTELLIGENCE_FACTS_UNSAFE" failed = (await db_session.execute( select(IntelligenceAnalysis).where(IntelligenceAnalysis.snapshot_id == snapshot_id) )).scalar_one() assert failed.status == "failed" assert failed.failure_code == "INTELLIGENCE_FACTS_UNSAFE" assert failed.input_token_count is None def test_the_seam_hands_the_model_a_flat_citable_package(): """Holds the WIRING, not just the helper. Testing ``_flatten_for_citation`` alone left this green when the call was removed from ``build_model_fact_package`` — the helper was correct and simply never invoked, which is the failure mode that matters. This goes through the seam every subject passes through, with a package shaped like the cloud subscription's (nested groups and lists) that production refused. """ from app.intelligence.contracts import _resolve_pointer snapshot = IntelligenceSnapshot( subject_type="cloud_scope", subject_id="real-scope-uuid", audience_key="shared", score_version="health-v1", facts_hash="b" * 64, overall_score=Decimal("70.00"), confidence_score=Decimal("90.00"), grade="healthy", dimensions=[{"key": "cloud_network", "score": 80}], aggregate_facts={ "costs": {"top_resources": [{"amount": 5}, {"amount": 3}]}, "data_quality": {"sources": [ {"available_count": 1}, {"available_count": 2}, {"available_count": 3}, {"available_count": 4}, ]}, "sync": {"lag_days": 4}, }, reference_map={"srv_01": {"kind": "server", "object_id": "private-uuid"}}, data_as_of=datetime(2026, 8, 17, tzinfo=UTC), ) package = build_model_fact_package(snapshot) assert all( not isinstance(value, (dict, list)) for value in package["facts"].values() ), "no nested value may remain for the model to mis-navigate" # The pointer production got wrong now resolves, as a scalar. assert _resolve_pointer( package, "/facts/data_quality.sources.2.available_count", ) == 3 assert _resolve_pointer(package, "/facts/sync.lag_days") == 4 assert _resolve_pointer(package, "/facts/costs.top_resources.count") == 2 # Every legal address is published as a copyable list, and every entry # in it resolves to a scalar. ``evidence_refs`` has always been # published this way and has never been invented; fact refs were left # to be COMPOSED from dict keys and were composed wrong — an # unresolvable path, then a missing prefix, then a wholly invented # ``.0.metrics.0.value``. The subject that works publishes 23 citable # keys; the one that kept failing publishes 124. refs = package["citable_fact_refs"] assert refs, "the legal addresses must be published, not inferred" for ref in refs: assert not isinstance( _resolve_pointer(package, ref), (dict, list) ), f"published ref must resolve to a scalar: {ref}" # Completeness is the point. Listing only /facts/* was the first # version, and production immediately showed the cost: the model wanted # the score, found no legal address for it, and invented # "/facts/overall_score" — where overall_score is a TOP-LEVEL member. assert "/overall_score" in refs assert "/grade" in refs assert "/confidence_score" in refs assert set(refs) >= {f"/facts/{key}" for key in package["facts"]} # Every scalar the model can see must be citable, or it will guess. scalars = { f"/{key}" for key, value in package.items() if not isinstance(value, (dict, list)) } assert scalars <= set(refs), f"unciteable scalars: {scalars - set(refs)}" # ...and the vocabularies must NOT be offered as facts to quote. # Enforced by the list rule: evidence_refs, horizon_days and the list # itself are all lists, and lists are not citable. assert not [r for r in refs if r.startswith(("/evidence_refs", "/horizon_days"))] @pytest.mark.asyncio async def test_cooldown_counts_failed_attempts_and_still_names_the_failure( db_engine, db_session, monkeypatch, ): """A deterministically-failing subject must not be retryable on a loop. Counting only ``pending|succeeded`` meant a caller whose model call kept being rejected (bad key, schema drift, grounding refusal) could re-fire immediately — each retry billed a real request — with nothing but the per-USER daily quota as a brake. The subject-level cooldown now counts the failure, and still tells the operator what broke. """ from app.core import database from app.intelligence import generation factory = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False) monkeypatch.setattr(database, "async_session_factory", factory) snapshot = _snapshot("cloud_scope", "scope-cooldown-failed", "e" * 64) db_session.add(snapshot) await db_session.flush() db_session.add(IntelligenceAnalysis( snapshot_id=snapshot.id, subject_type=snapshot.subject_type, subject_id=snapshot.subject_id, audience_key=snapshot.audience_key, analysis_kind="cloud_scope", facts_hash=snapshot.facts_hash, generation_key=uuid.uuid4().hex, status="failed", failure_code="INTELLIGENCE_OUTPUT_INVALID", failure_message="Model output failed grounding or schema validation", data_as_of=snapshot.data_as_of, generated_at=datetime.now(UTC), )) await db_session.commit() class MustNotRun: def __init__(self, *args, **kwargs): raise AssertionError("cooldown must fire before any billed call") monkeypatch.setattr(generation, "BailianLLMClient", MustNotRun) with pytest.raises(AppException) as caught: await generate_analysis( db_session, snapshot=snapshot, analysis_kind="cloud_scope", actor_id=None, enforce_user_quota=False, ) assert caught.value.code == "INTELLIGENCE_COOLDOWN" assert caught.value.status_code == 429 # The throttle must not swallow WHY the previous run failed, or it reads # as Helix being arbitrary. Only the stable code travels — the # model-adjacent ``failure_message`` stays behind the privileged view. assert "INTELLIGENCE_OUTPUT_INVALID" in caught.value.message assert "grounding or schema validation" not in caught.value.message assert await db_session.scalar( select(func.count()).select_from(IntelligenceAnalysis) ) == 1 @pytest.mark.asyncio async def test_cache_hit_marks_the_served_analysis_row_as_reused( db_engine, db_session, monkeypatch, ): """``cache_reused`` must be true on the row, not only in the request. ``read_latest_analysis`` and the subscription projection both read the persisted column, but nothing ever wrote True — so a brief that had not been regenerated for days reported ``cache_reused: false`` forever. """ from app.core import database from app.intelligence import generation factory = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False) monkeypatch.setattr(database, "async_session_factory", factory) snapshot = _snapshot("cloud_scope", "scope-cache-flag", "d" * 64) snapshot.aggregate_facts = {"month_cost": 1200, "budget_utilization": 80} snapshot.reference_map = {"scope_01": {"kind": "scope", "object_id": "local-id"}} db_session.add_all([ snapshot, PlatformSetting(key="ai_spending.llm.api_key", value="test-key"), ]) await db_session.commit() class FakeClient: def __init__(self, *args, **kwargs): pass async def chat_json(self, **kwargs): return { **_valid_output(), "_meta": { "model_used": "fake-grounded-model", "input_tokens": 12, "output_tokens": 34, }, } monkeypatch.setattr(generation, "BailianLLMClient", FakeClient) first = await generate_analysis( db_session, snapshot=snapshot, analysis_kind="cloud_scope", actor_id=None, enforce_user_quota=False, ) assert first.cache_reused is False assert first.analysis.cache_reused is False second = await generate_analysis( db_session, snapshot=snapshot, analysis_kind="cloud_scope", actor_id=None, enforce_user_quota=False, ) assert second.cache_reused is True assert second.analysis.id == first.analysis.id stored = await db_session.get( IntelligenceAnalysis, first.analysis.id, populate_existing=True, ) assert stored.cache_reused is True def test_the_derived_reference_list_does_not_trip_the_privacy_scan(): """Real production data caught this; no unit fixture would have. Publishing ``citable_fact_refs`` BEFORE ``assert_llm_safe`` promoted fact key names into scanned values, and dict keys were never scanned here. The cloud subscription package was then rejected outright with "domain-like value" on eight of its own reference strings — ``/dimensions/0.score`` and friends read as hostnames — even though nothing unsafe had entered the payload. Changing the separator was the wrong fix: the dot is load-bearing in the resolver's tolerant spellings. Scanning real content first and attaching the derived vocabulary after restores the original values-only boundary. """ snapshot = IntelligenceSnapshot( subject_type="cloud_scope", subject_id="real-scope-uuid", audience_key="shared", score_version="health-v1", facts_hash="c" * 64, overall_score=Decimal("70.00"), confidence_score=Decimal("90.00"), grade="healthy", # The real production shape, taken from the rejected package: a # metric named ``cloud.access.governed_coverage`` under ``metrics`` # flattens to ``...metrics.cloud.access.governed_coverage``, and # ``metrics.cloud`` matches the privacy scanner's domain pattern # because ``cloud`` is one of its TLDs. 157 of that package's own # reference strings matched this way. dimensions=[{ "key": "access_governance", "score": 80, "metrics": [ {"key": "cloud.access.governed_coverage", "value": 42}, ], }], aggregate_facts={"costs": {"top_resources": [{"amount": 5}]}}, reference_map={}, data_as_of=datetime(2026, 8, 17, tzinfo=UTC), ) package = build_model_fact_package(snapshot) assert package["citable_fact_refs"], "the vocabulary must still be published" from app.intelligence.privacy import _DOMAIN flagged = [ ref for ref in package["citable_fact_refs"] if _DOMAIN.search(ref) ] assert flagged, ( "this test is only meaningful while it publishes an address the " "scanner would reject; production had 157 of them" )