当互联网开始由AI书写：2026企业级AI合成数据污染、递归自噬、幻影共识与来源断链治理实战

┌────────────────────────────────────────────────────────────────────────────────┐
│  2026 Synthetic Contamination & Corpus Commons Decay: Five-Layer Model         │
├────────────────────────────────────────────────────────────────────────────────┤
│                                                                                │
│  [Zero-cost generation + Unmeasured verification + Circular retrieval +        │
│   Volatile provenance marks → AI output feeds AI input; tails vanish first;    │
│   echoes masquerade as consensus; originals disappear; the commons floods      │
│   until relevance-ranking returns noise by construction]                       │
│       ↓                                                                        │
│  ┌─ L1: 来源血缘登记层 (Provenance Lineage Registry) ────────────────────────┐ │
│  │  • 生成即登记: AI产出物携带三重血缘(生成模型/时间戳/内容指纹)方可入库      │ │
│  │  • 无源拒收: 无法出示血缘的内容按污染内容处置, 禁止进入知识与训练语料      │ │
│  │  • 跨平台存活: 血缘嵌入语义指纹而非文件元数据, 抵抗格式转换洗白            │ │
│  │  • 原件绑定: AI摘要/改写必须绑定原件哈希, 原件删除即摘要失效               │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L2: 合成/人类语料分池层 (Synthetic-Human Corpus Partitioning) ───────────┐ │
│  │  • 双池强制: 知识语料分人类池与合成池, 检索与训练分别声明配比              │ │
│  │  • 累积非替换: 合成内容禁止挤占人类内容的存储与检索权重(替换即崩塌)        │ │
│  │  • 占比普查: 季度合成占比审计, 未测量即视为高危                           │ │
│  │  • 尾部保护: 人类池中罕见/长尾内容设删除豁免, 禁止按"低访问"清理           │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L3: 引用环度与幻影共识检测层 (Circularity & Phantom Consensus Detection)─┐ │
│  │  • 引用图审计: 文档引用图的强连通分量检测, 闭环引用即冻结展开              │ │
│  │  • 独立性折算: 多来源一致的置信度按来源间信息独立性折算, 同源回声不计票     │ │
│  │  • 上游溯源: 每条"事实"可回溯至引用链终点, 终点必须是现实锚而非生成物       │ │
│  │  • 幻觉入库阻断: AI生成内容禁止直接成为下一轮grounding的权威来源           │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L4: 现实锚点与崩塌熔断层 (Reality Anchoring & Collapse Circuit Breaking)─┐ │
│  │  • 锚点配额: 关键域知识必须保有最低比例的现实锚(原件/实测/人工确认)        │ │
│  │  • 尾部监测: 语料与输出的分布尾部覆盖率/方差季度审计, 收缩即冻结合成入库    │ │
│  │  • 抽样现实核对: 检索高频事实定期与物理世界/原始系统抽样比对               │ │
│  │  • 原件保全令: 高影响域(医药/金融/法务/安全)原件禁止删除, 存储成本强制预算  │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L5: 公地排放责任层 (Commons Emission Accountability) ────────────────────┐ │
│  │  • 排放核算: AI内容发布量计入"语料公地成本", 无标记排放限流与追责           │ │
│  │  • 污染损失归因: 幻影共识/循环引用导致的决策损失归入生成系统生态成本        │ │
│  │  • 行业协同: 跨企业来源互认与黑名单指纹共享                               │ │
│  │  • 架构归责: "把记录的转述当作现实的记录治理"的责任框架                    │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│                                                                                │
└────────────────────────────────────────────────────────────────────────────────┘

"""
provenance_circularity_audit.py - 来源血缘与引用环度审计引擎
核心原则: "知识库里有这条内容"不等于"这条内容有来源"——
          如果AI摘要入库后原件被删除,
          那循环里再也没有一个点能被现实校准,
          知识库从现实的缓存变成了现实的替代品;
          如果14份文档互相引用一个从未存在的Nature结论,
          那不是14条证据, 是同一个幻觉的14次复印,
          而复印比原件更干净——因为每一代转述
          都磨掉了上一代的犹豫;
          如果七家机构"一致看好"的源头是同一个段落,
          那一致性不是共识, 是回声——
          而回声恰好骗过所有为检测谎言设计的系统,
          因为它在形式上就是交叉验证的样子;
          血缘审计最反直觉的地方在于:
          你要找的不是"哪条内容是假的",
          而是"哪条内容的引用链, 最终没有落在现实上"——
          因为递归自噬的可怕不在于错误,
          在于正确与错误在循环闭合之后,
          失去了被区分的接口
"""
from typing import Dict, List, Any, Optional, Tuple, Set
from enum import Enum
from dataclasses import dataclass, field
from collections import defaultdict, deque
import time, uuid, json, hashlib, re
import numpy as np

class ContentOrigin(str, Enum): jiyi.tongsou.com 
    HUMAN_AUTHORED = "human"          # 人类原创(现实锚)
    AI_GENERATED = "ai_generated"     # AI生成(须带血缘)
    AI_ASSISTED = "ai_assisted"       # AI辅助人类定稿(混合)
    AI_SUMMARY = "ai_summary"         # AI摘要(须绑定原件)
    UNKNOWN = "unknown"               # 无源(视同污染)

class ProvenanceState(str, Enum):
    REGISTERED = "registered"        # 血缘完整
    ORPHANED = "orphaned"            # 原件缺失(摘要失效)
    CIRCULAR = "circular"            # 引用闭环
    UNANCHORED = "unanchored"        # 引用链终点非现实锚
    REJECTED = "rejected"            # 拒收

class ContaminationPattern(str, Enum):
    RECURSIVE_CONSUMPTION = "recursive"     # 递归自噬
    PHANTOM_CONSENSUS = "phantom_consensus" # 幻影共识
    CIRCULAR_CITATION = "circular_citation" # 闭环引用
    PROVENANCE_LAUUNDERING = "laundering"   # 现实洗白
    ORIGINAL_DELETION = "original_deletion" # 原件消失
    TAIL_COLLAPSE = "tail_collapse"         # 尾部坍塌
    NONE = "none"

@dataclass
class ContentLineage: zhaixing.tongsou.com 
    """内容血缘登记——来源是内容身份的一部分"""
    content_id: str = field(default_factory=lambda: f"cl-{uuid.uuid4().hex[:10]}")
    corpus_id: str = ""
    origin: ContentOrigin = ContentOrigin.UNKNOWN
    # 三重血缘
    generator_model: str = ""          # 生成模型(若AI)
    generated_at: float = 0.0
    semantic_fingerprint: str = ""     # 语义指纹(抗格式转换)
    # 摘要绑定
    original_content_id: str = ""      # 原件ID(摘要类必填)
    original_hash: str = ""
    original_exists: bool = True
    # 引用
    cites: List[str] = field(default_factory=list)      # 本文引用的内容ID
    cited_by: List[str] = field(default_factory=list)
    # 现实锚
    reality_anchor: str = ""           # 实验/原件/人工确认/物理记录
    # 状态
    state: ProvenanceState = ProvenanceState.REGISTERED
    ingested_at: float = field(default_factory=time.time)

@dataclass
class CorpusCensus: xunling.tongsou.com 
    """语料库普查"""
    census_id: str = field(default_factory=lambda: f"cc-{uuid.uuid4().hex[:10]}")
    corpus_id: str = ""
    total_documents: int = 0
    synthetic_share: float = 0.0        # 合成内容占比
    unregistered_share: float = 0.0     # 无源占比
    orphaned_summary_share: float = 0.0 # 原件缺失的摘要占比
    tail_coverage: float = 0.0          # 分布尾部覆盖率
    tail_trend: str = ""                # "stable" / "shrinking" / "collapsing"
    anchored_fact_share: float = 0.0    # 事实类内容中锚定现实的比例
    censused_at: float = field(default_factory=time.time)

@dataclass
class CircularityFinding:
    """环度审计发现"""
    finding_id: str = field(default_factory=lambda: f"cf-{uuid.uuid4().hex[:10]}")
    corpus_id: str = ""
    scc_size: int = 0                   # 强连通分量规模
    scc_members: List[str] = field(default_factory=list)
    entry_point: str = ""               # 幻觉/污染进入循环的节点
    generations_deep: int = 0           # 引用加固代数
    found_at: float = field(default_factory=time.time)

class ProvenanceCircularityEngine: maifushi.tongsou.com 
    """来源血缘与引用环度审计引擎"""

    # 配置
    UNKNOWN_ORIGIN_REJECT = True          # 无源拒收
    SUMMARY_ORIGINAL_MANDATORY = True     # 摘要必须绑定原件
    SYNTHETIC_SHARE_ALERT = 0.50          # 合成占比>50%高危
    UNREGISTERED_SHARE_ALERT = 0.10       # 无源占比>10%告警
    TAIL_SHRINK_ALERT = 0.15              # 尾部覆盖率季度收缩>15%熔断
    ANCHORED_FACT_FLOOR = 0.70            # 事实类内容现实锚定率下限
    SCC_SIZE_ALERT = 3                    # 引用闭环≥3节点即冻结
    PHANTOM_CONSENSUS_MIN_SOURCES = 3     # ≥3"独立"来源同源即幻影共识
    HIGH_IMPACT_NO_DELETE = ("medical", "regulatory", "financial",
                             "legal", "safety", "security")

    def __init__(self, app_registry, alerts, audit, metrics):
        self.registry = app_registry
        self.alerts = alerts
        self.audit = audit
        self.metrics = metrics
        self.contents: Dict[str, ContentLineage] = {}
        self.by_corpus: Dict[str, List[ContentLineage]] = defaultdict(list)
        self.censuses: List[CorpusCensus] = []
        self.findings: List[CircularityFinding] = []

    @staticmethod
    def _fp(text: str) -> str:
        return hashlib.sha256(text.encode("utf-8")).hexdigest()[:24]

    async def gate_ingestion(self, corpus_id: str, candidate: ContentLineage
                              ) -> Dict[str, Any]:
        """入库总闸: 血缘完整性+原件绑定+无源拒收"""

        violations = []

        # 规则1: 来源必须声明
        if candidate.origin == ContentOrigin.UNKNOWN:
            if self.UNKNOWN_ORIGIN_REJECT: zhendao.tongsou.com 
                violations.append("UNKNOWN ORIGIN — no lineage presented; "
                                  "content without provenance is not "
                                  "'unmarked', it is unaccountable")
        # 规则2: AI生成必须携带生成血缘
        elif candidate.origin == ContentOrigin.AI_GENERATED and \
             not candidate.generator_model:
            violations.append("AI CONTENT WITHOUT GENERATOR IDENTITY")

        # 规则3: 摘要必须绑定仍存在的原件
        if candidate.origin == ContentOrigin.AI_SUMMARY:
            if not candidate.original_content_id or not candidate.original_hash:
                violations.append("SUMMARY WITHOUT ORIGINAL BINDING")
            elif not candidate.original_exists: weimeng.tongsou.com 
                candidate.state = ProvenanceState.ORPHANED
                violations.append(
                    f"ORIGINAL DELETED — summary for "
                    f"'{candidate.original_content_id}' is now an "
                    f"unfalsifiable transcription of a destroyed source")

        # 规则4: 事实类声明必须有现实锚
        if candidate.reality_anchor == "" and \
           candidate.origin in (ContentOrigin.AI_GENERATED,
                                ContentOrigin.AI_SUMMARY):
            violations.append("NO REALITY ANCHOR — generative content "
                              "asserting facts must name an experiment, "
                              "record, measurement or human confirmation")

        result = {"allowed": not violations, "violations": violations,
                  "content_id": candidate.content_id}

        if violations: aisou.tongsou.com 
            candidate.state = ProvenanceState.REJECTED
            await self.alerts.critical(
                f"🚫 INGESTION REJECTED: Corpus '{corpus_id}', content "
                f"'{candidate.content_id}' ({candidate.origin.value}). "
                f"Violations: {violations[:3]}. "
                f"Every recursive loop starts with one unexamined "
                f"acceptance: a hallucinated citation filed as "
                f"'literature the AI found', a summary kept after the "
                f"original was deleted for storage savings, a generated "
                f"assertion with no experiment behind it. "
                f"None of them looks dangerous on the day it enters. "
                f"The pharma knowledge base that reached a regulator "
                f"with a 14-generation phantom citation started as one "
                f"of these— and the gate that should have asked "
                f"'where does this end?' was never built."
            )
        else: toujing.tongsou.com 
            self.contents[candidate.content_id] = candidate
            self.by_corpus[corpus_id].append(candidate)

        await self.audit.log_ingestion_gate(candidate, result)
        return result

    async def census_corpus(self, corpus_id: str,
                             tail_measurements: Dict[str, float]
                             ) -> CorpusCensus:
        """语料库季度普查: 合成占比+无源占比+尾部+锚定率"""

        docs = self.by_corpus.get(corpus_id, [])
        if not docs:
            return CorpusCensus(corpus_id=corpus_id)

        n = len(docs)
        synthetic = sum(1 for d in docs if d.origin in
                        (ContentOrigin.AI_GENERATED, ContentOrigin.AI_SUMMARY))
        unregistered = sum(1 for d in docs
                           if d.origin == ContentOrigin.UNKNOWN
                           or d.state == ProvenanceState.REJECTED)
        orphaned = sum(1 for d in docs
                       if d.origin == ContentOrigin.AI_SUMMARY
                       and not d.original_exists)
        fact_docs = [d for d in docs if d.origin != ContentOrigin.HUMAN_AUTHORED]
        anchored = (sum(1 for d in fact_docs if d.reality_anchor)
                    / max(len(fact_docs), 1))

        census = CorpusCensus(
            corpus_id=corpus_id, total_documents=n,
            synthetic_share=synthetic / n,
            unregistered_share=unregistered / n,
            orphaned_summary_share=orphaned / n,
            tail_coverage=tail_measurements.get("tail_coverage", 1.0),
            anchored_fact_share=anchored)

        # 尾部趋势: 与上次普查比对
        prev = [c for c in self.censuses if c.corpus_id == corpus_id]
        if prev:
            shrink = prev[-1].tail_coverage - census.tail_coverage
            if shrink > self.TAIL_SHRINK_ALERT:
                census.tail_trend = "collapsing"
            elif shrink > 0.05:
                census.tail_trend = "shrinking"
            else:
                census.tail_trend = "stable"

        self.censuses.append(census)

        # 分级告警
        if census.synthetic_share > self.SYNTHETIC_SHARE_ALERT:
            await self.alerts.critical(
                f"🚨 CORPUS COMPOSITION CRITICAL: '{corpus_id}'. "
                f"Synthetic share: {census.synthetic_share:.0%} "
                f"({synthetic}/{n}). "
                f"This corpus no longer describes your organization— "
                f"it describes what your organization's models "
                f"previously said about your organization. "
                f"The question 'what do we know?' and the question "
                f"'what did we generate?' have merged, and no "
                f"retrieval system can tell them apart anymore."
            )
        if census.unregistered_share > self.UNREGISTERED_SHARE_ALERT:
            await self.alerts.critical(
                f"🚨 UNPROVENANCED CONTENT: '{corpus_id}'. "
                f"{census.unregistered_share:.0%} of documents carry no "
                f"lineage. Unmeasured is not neutral: in a corpus where "
                f"60% of new content is machine-made, 'unknown origin' "
                f"is a synthetic confession by base rate."
            )
        if census.orphaned_summary_share > 0.20:
            await self.alerts.critical(
                f"🚨 ORPHANED SUMMARIES: '{corpus_id}'. "
                f"{census.orphaned_summary_share:.0%} of summaries have "
                f"no surviving original. "
                f"41% of enterprises deleted the source after keeping "
                f"the summary— and discovered, in the audit that "
                f"mattered, that a summary without an original is not "
                f"knowledge. It is an unfalsifiable claim wearing "
                f"knowledge's clothes. The storage you saved cost "
                f"4.7 hundred million at the moment a regulator asked "
                f"for the paper that no longer existed."
            )
        if census.anchored_fact_share < self.ANCHORED_FACT_FLOOR:
            await self.alerts.critical(
                f"🚨 REALITY ANCHOR DEFICIT: '{corpus_id}'. "
                f"Only {census.anchored_fact_share:.0%} of generative "
                f"fact-claims name an anchor (floor "
                f"{self.ANCHORED_FACT_FLOOR:.0%}). "
                f"A corpus is a map. Anchors are the survey points "
                f"that tie the map to the territory. "
                f"Without them, every update is drawn from the "
                f"previous map— and the map drifts from the territory "
                f"at a rate no one inside the map can measure."
            )
        if census.tail_trend == "collapsing": qiyin.tongsou.com 
            await self.alerts.critical(
                f"🛑 TAIL COLLAPSE DETECTED: '{corpus_id}'. "
                f"Tail coverage fell >{self.TAIL_SHRINK_ALERT:.0%} "
                f"since last census ({census.tail_coverage:.2f}). "
                f"FREEZE all synthetic ingestion immediately. "
                f"Model collapse does not announce itself as an error "
                f"rate— it announces itself as a SILENCE: rare cases, "
                f"edge knowledge, minority patterns stop appearing, "
                f"and because the mean got smoother, everything looks "
                f"better right up until the first edge case arrives "
                f"and finds that the organization deleted the only "
                f"copy of itself that knew how to handle it."
            )

        await self.audit.log_corpus_census(census)
        return census

    async def audit_citation_graph(self, corpus_id: str
                                    ) -> List[CircularityFinding]:
        """引用图环度审计: 强连通分量检测(Tarjan)"""

        docs = {d.content_id: d for d in self.by_corpus.get(corpus_id, [])}

        # Tarjan SCC
        index_counter = [0]
        stack, lowlink, index, on_stack = [], {}, {}, set()
        sccs = []

        def strongconnect(v):
            index[v] = index_counter[0]
            lowlink[v] = index_counter[0]
            index_counter[0] += 1
            stack.append(v)
            on_stack.add(v)
            for w in docs.get(v, ContentLineage()).cites:
                if w not in docs:
                    continue
                if w not in index:
                    strongconnect(w)
                    lowlink[v] = min(lowlink[v], lowlink[w])
                elif w in on_stack:
                    lowlink[v] = min(lowlink[v], index[w])
            if lowlink[v] == index[v]:
                comp = []
                while True:
                    w = stack.pop()
                    on_stack.discard(w)
                    comp.append(w)
                    if w == v:
                        break
                if len(comp) > 1 or v in docs.get(v, ContentLineage()).cites:
                    sccs.append(comp)

        for v in list(docs.keys()):
            if v not in index:
                strongconnect(v)

        findings = []
        for scc in sccs:
            if len(scc) >= self.SCC_SIZE_ALERT:
                # 寻找环内最可能的污染入口: 无外部锚且最古老
                entry = min(scc, key=lambda cid: (
                    bool(docs[cid].reality_anchor),
                    docs[cid].ingested_at))
                finding = CircularityFinding(
                    corpus_id=corpus_id, scc_size=len(scc),
                    scc_members=scc[:20], entry_point=entry,
                    generations_deep=self._max_chain_depth(scc, docs))
                findings.append(finding)
                self.findings.append(finding)

                for cid in scc:
                    docs[cid].state = ProvenanceState.CIRCULAR

                await self.alerts.critical(
                    f"🚨 CIRCULAR CITATION RING: Corpus '{corpus_id}'. "
                    f"{len(scc)} documents in a closed citation loop. "
                    f"Probable entry point: '{entry}' "
                    f"(anchor: '{docs[entry].reality_anchor or 'NONE'}'). "
                    f"Chain depth: {finding.generations_deep} generations. "
                    f"ALL MEMBERS FROZEN pending lineage unwind. "
                    f"Fourteen documents cited each other around a "
                    f"Nature paper that was never published. "
                    f"Each generation of retelling was CLEANER than "
                    f"the last— hedging removed, confidence sharpened, "
                    f"the hallucination polished into doctrine. "
                    f"A citation ring is not evidence. "
                    f"It is one claim, photocopying itself, "
                    f"and every copy loses a little more "
                    f"of the original's doubt."
                )

        await self.audit.log_citation_audit(corpus_id, findings)
        return findings

    def _max_chain_depth(self, scc: List[str],
                          docs: Dict[str, ContentLineage]) -> int:
        """环内最大引用代数(近似: 按入库时间分层)"""
        ordered = sorted(scc, key=lambda c: docs[c].ingested_at)
        depth, seen = 0, set()
        for cid in ordered:
            for w in docs[cid].cites:
                if w in docs and w not in seen:
                    depth += 1
                    seen.add(w)
        return min(depth, len(scc))

    async def detect_phantom_consensus(self, claim_key: str,
                                        sources: List[Dict[str, Any]]
                                        ) -> Optional[Dict[str, Any]]:
        """幻影共识检测: 多来源一致但语义指纹同源
        sources: [{"source_name":..., "content_id":...,
                   "semantic_fingerprint":...}]"""

        if len(sources) < self.PHANTOM_CONSENSUS_MIN_SOURCES:
            return None

        # 指纹聚类: 相似度高于阈值视为同源
        clusters = self._cluster_fingerprints(sources)
        largest = max(clusters, key=len)

        if len(largest) >= self.PHANTOM_CONSENSUS_MIN_SOURCES and \
           len(largest) == len(sources):
            result = {
                "claim_key": claim_key,
                "verdict": "PHANTOM_CONSENSUS",
                "apparent_sources": len(sources),
                "independent_sources": 1,
                "shared_fingerprint_cluster": [s["source_name"]
                                                for s in largest]
            }
            await self.alerts.critical(
                f"🚨 PHANTOM CONSENSUS: Claim '{claim_key}' appears "
                f"consistent across {len(sources)} 'independent' "
                f"sources— all trace to ONE semantic fingerprint. "
                f"Independent sources: 1. "
                f"Seven research desks, one paragraph, 23 billion in, "
                f"9.2 billion out. Consensus derives its evidential "
                f"power from INDEPENDENCE— when N observers drink from "
                f"the same well, their agreement is not corroboration, "
                f"it is the absence of sampling noise. "
                f"Worse: agreement-of-echoes defeats every deception "
                f"detector ever built, because it is structurally "
                f"indistinguishable from cross-validation. "
                f"Count wells, not voices."
            )
            await self.audit.log_phantom_consensus(result)
            return result

        return {"claim_key": claim_key, "verdict": "genuine_pluralism",
                "independent_clusters": len(clusters)}

    def _cluster_fingerprints(self, sources: List[Dict[str, Any]]
                               ) -> List[List[Dict[str, Any]]]:
        """语义指纹聚类(简化: 哈希前缀分桶, 生产接嵌入相似度)"""
        buckets = defaultdict(list)
        for s in sources:
            fp = s.get("semantic_fingerprint", "")
            buckets[fp[:8]].append(s)   # 前8位相同视为同源(示意)
        return list(buckets.values())

    async def verify_original_binding(self, summary_id: str
                                       ) -> Dict[str, Any]:
        """摘要-原件绑定核验(原件删除即摘要失效)"""

        summary = self.contents.get(summary_id)
        if not summary or summary.origin != ContentOrigin.AI_SUMMARY:
            return {"error": "not a registered summary"}

        original = self.contents.get(summary.original_content_id)
        if not original:
            summary.original_exists = False
            summary.state = ProvenanceState.ORPHANED
            await self.alerts.critical(
                f"🚨 SUMMARY ORPHANED: '{summary_id}' references deleted "
                f"original '{summary.original_content_id}'. "
                f"Summary INVALIDATED — retrieval weight zeroed. "
                f"The on-call engineer followed the highest-voted "
                f"'recovery procedure' in the knowledge base for four "
                f"hours of outage, and the only document that could "
                f"have revealed the procedure was eight months "
                f"obsolete had been archived to save storage. "
                f"A summary is a lossy projection of its original. "
                f"Delete the original and the projection becomes "
                f"unfalsifiable— and unfalsifiable content in a "
                f"knowledge base is not knowledge. It is a belief "
                f"with a timestamp."
            )
            await self.audit.log_orphaned_summary(summary_id)
            return {"valid": False, "state": "orphaned"}

        # 哈希一致性
        if original.semantic_fingerprint != summary.original_hash:
            await self.alerts.warning(
                f"⚠️ SUMMARY DRIFT: '{summary_id}' bound hash does not "
                f"match current original — original was revised after "
                f"summarization. Re-summarize or invalidate."
            )
            return {"valid": False, "state": "stale_binding"}

        return {"valid": True, "state": "bound"}

"""
corpus_quarantine_defense.py - 语料分池与崩塌熔断防御引擎
核心: 模型崩塌不是训练的意外, 是替换制度的必然输出——
      Gerstgrasser原理: 累积制下合成数据无害甚至有益,
      替换制下任何比例的合成数据都在启动尾部消失的倒计时;
      你不能通过"更努力地生成高质量内容"来防崩塌,
      因为崩塌的机制恰恰是质量: 每一代生成都更流畅、
      更标准、更向均值收敛——均值附近越好, 尾部死得越快;
      你能做的是制度性隔断: 双池分治让人类语料的权重
      不随合成内容的产量贬值, 尾部豁免让罕见知识
      不因低访问被清理, 现实抽样让高频"事实"定期
      在物理世界遇到一次对照, 崩塌熔断让方差收缩
      在变成组织失忆之前冻结入库;
      防御的本质不是控制生成的量, 是保住循环里的锚——
      而在一个内容边际成本为零的时代,
      最贵的从来不是存储原件的那点磁盘,
      是承认这句话: 一个不留原件的知识库,
      保存的不是知识, 是知识的传闻
"""
from typing import Dict, List, Any, Optional, Tuple
from enum import Enum
from dataclasses import dataclass, field
from collections import defaultdict
import time, uuid, json
import numpy as np

class PoolType(str, Enum):
    HUMAN = "human_pool"          # 人类池(现实锚, 权重保护)
    SYNTHETIC = "synthetic_pool"  # 合成池(标注, 限额, 可清除)
    QUARANTINE = "quarantine"     # 隔离池(污染嫌疑, 禁止检索)

class CollapseSignal(str, Enum):
    TAIL_SHRINK = "tail_shrink"          # 尾部覆盖率收缩
    VARIANCE_CONTRACTION = "variance"    # 输出方差收缩
    DIVERSITY_LOSS = "diversity"         # 主题多样性下降
    ANCHOR_DEFICIT = "anchor_deficit"    # 现实锚定率不足
    REPLACEMENT_DETECTED = "replacement" # 替换制度检出

class CircuitState(str, Enum):
    CLOSED = "closed"            # 正常
    THROTTLED = "throttled"      # 限流(合成入库降速)
    FROZEN = "frozen"            # 熔断(合成入库全停)

@dataclass
class PoolPartition:
    """双池分区"""
    corpus_id: str = ""
    human_pool_docs: int = 0
    synthetic_pool_docs: int = 0
    quarantine_docs: int = 0
    # 检索权重配比(制度性保护)
    human_retrieval_weight: float = 0.7
    synthetic_retrieval_weight: float = 0.3
    # 替换检测: 人类池是否在收缩
    human_pool_delta_qoq: float = 0.0     # 人类池季度变化
    replacement_regime: bool = False       # 检出替换制度
    partitioned_at: float = field(default_factory=time.time)

@dataclass
class CollapseMonitor:
    """崩塌监测快照"""
    monitor_id: str = field(default_factory=lambda: f"cm-{uuid.uuid4().hex[:10]}")
    corpus_id: str = ""
    tail_coverage: float = 0.0        # 尾部覆盖率
    output_variance: float = 0.0      # 输出多样性方差
    topic_diversity: float = 0.0      # 主题熵
    signals: List[CollapseSignal] = field(default_factory=list)
    circuit: CircuitState = CircuitState.CLOSED
    sampled_at: float = field(default_factory=time.time)

@dataclass
class RealityCheck:
    """现实抽样核对"""
    check_id: str = field(default_factory=lambda: f"rc-{uuid.uuid4().hex[:10]}")
    corpus_id: str = ""
    facts_sampled: int = 0
    facts_verified_against_reality: int = 0
    facts_diverged: int = 0
    divergence_rate: float = 0.0
    diverged_items: List[Dict[str, Any]] = field(default_factory=list)
    checked_at: float = field(default_factory=time.time)

@dataclass
class EmissionRecord:
    """公地排放核算"""
    org_id: str = ""
    period: str = ""
    ai_content_published: int = 0
    labeled_share: float = 0.0
    commons_cost_credits: float = 0.0
    unlabeled_penalty: float = 0.0

class CorpusQuarantineEngine: hanzhi.tongsou.com 
    """语料分池与崩塌熔断防御引擎"""

    # 配置
    HUMAN_WEIGHT_FLOOR = 0.60            # 人类池检索权重下限
    REPLACEMENT_ALERT = -0.05            # 人类池季度收缩>5%即替换嫌疑
    TAIL_FLOOR = 0.30                    # 尾部覆盖率绝对下限
    VARIANCE_SHRINK_ALERT = 0.20         # 方差季度收缩>20%限流
    VARIANCE_SHRINK_FREEZE = 0.35        # >35%熔断
    REALITY_CHECK_DIVERGENCE_ALERT = 0.10 # 现实核对偏离率>10%冻结相关域
    REALITY_CHECK_SAMPLE_PCT = 2.0       # 高频事实抽样比例
    TAIL_DELETE_EXEMPTION = True         # 长尾内容删除豁免
    HIGH_IMPACT_ORIGINAL_IMMORTAL = True # 高影响域原件永久保全

    def __init__(self, provenance_engine, alerts, audit, metrics):
        self.provenance = provenance_engine
        self.alerts = alerts
        self.audit = audit
        self.metrics = metrics
        self.partitions: Dict[str, PoolPartition] = {}
        self.monitors: Dict[str, List[CollapseMonitor]] = defaultdict(list)
        self.circuit_states: Dict[str, CircuitState] = {}
        self.emissions: List[EmissionRecord] = []

    async def enforce_partition(self, corpus_id: str,
                                 partition: PoolPartition
                                 ) -> PoolPartition:
        """双池分治: 权重保护与替换检测"""

        self.partitions[corpus_id] = partition

        violations = []

        # 权重下限
        if partition.human_retrieval_weight < self.HUMAN_WEIGHT_FLOOR:
            violations.append(
                f"human pool retrieval weight "
                f"{partition.human_retrieval_weight:.0%} < floor "
                f"{self.HUMAN_WEIGHT_FLOOR:.0%}")

        # 替换制度检测: 人类池绝对量收缩而合成池增长
        if partition.human_pool_delta_qoq < self.REPLACEMENT_ALERT:
            partition.replacement_regime = True
            violations.append(
                f"REPLACEMENT REGIME DETECTED — human pool shrank "
                f"{partition.human_pool_delta_qoq:.0%} QoQ while "
                f"synthetic pool grew; this is the exact mechanism "
                f"Strong Model Collapse identified: substitution, "
                f"not proportion")

        if violations:
            await self.alerts.critical(
                f"🚨 PARTITION VIOLATION: Corpus '{corpus_id}'. "
                f"{' ; '.join(violations)}. "
                f"Gerstgrasser's finding is a gift and nobody reads "
                f"it: ACCUMULATE real and synthetic data together and "
                f"collapse does not occur; REPLACE real with synthetic "
                f"and collapse occurs at ANY proportion. "
                f"The danger was never 'too much AI content'— it is "
                f"'AI content instead of human content'. "
                f"Every storage-optimization job that prunes 'low "
                f"traffic' human documents to make room for fresh "
                f"generated ones is not cleaning the corpus. "
                f"It is winding the countdown."
            )

        await self.audit.log_partition(partition)
        return partition

    async def monitor_collapse(self, corpus_id: str,
                                snapshot: CollapseMonitor
                                ) -> CollapseMonitor:
        """崩塌监测与熔断"""

        snapshot.corpus_id = corpus_id
        history = self.monitors[corpus_id]
        history.append(snapshot)

        signals = []

        # 尾部绝对下限
        if snapshot.tail_coverage < self.TAIL_FLOOR:
            signals.append(CollapseSignal.TAIL_SHRINK)

        # 方差收缩(与上季度比)
        if len(history) >= 2:
            prev = history[-2]
            if prev.output_variance > 0:
                shrink = 1 - snapshot.output_variance / prev.output_variance
                if shrink > self.VARIANCE_SHRINK_FREEZE:
                    signals.append(CollapseSignal.VARIANCE_CONTRACTION)
                    snapshot.circuit = CircuitState.FROZEN
                elif shrink > self.VARIANCE_SHRINK_ALERT:
                    signals.append(CollapseSignal.VARIANCE_CONTRACTION)
                    snapshot.circuit = CircuitState.THROTTLED

        # 主题多样性
        if len(history) >= 2 and snapshot.topic_diversity < \
           history[-2].topic_diversity * 0.85:
            signals.append(CollapseSignal.DIVERSITY_LOSS)

        snapshot.signals = signals
        self.circuit_states[corpus_id] = snapshot.circuit

        if snapshot.circuit != CircuitState.CLOSED:
            verb = "FROZEN — all synthetic ingestion halted" if \
                   snapshot.circuit == CircuitState.FROZEN else \
                   "THROTTLED — synthetic ingestion rate-limited"
            await self.alerts.critical(
                f"🛑 COLLAPSE CIRCUIT {verb}: Corpus '{corpus_id}'. "
                f"Signals: {[s.value for s in signals]}. "
                f"Tail coverage: {snapshot.tail_coverage:.2f} "
                f"(floor {self.TAIL_FLOOR}). "
                f"Variance: {snapshot.output_variance:.3f}. "
                f"Topic entropy: {snapshot.topic_diversity:.3f}. "
                f"Collapse arrives as improvement: outputs get "
                f"smoother, more consistent, more standard— every "
                f"quality metric that measures the MEAN goes up while "
                f"the distribution loses its edges. "
                f"By the time the first rare case arrives and the "
                f"organization discovers it deleted the only copy of "
                f"itself that could handle it, the tails have been "
                f"gone for six quarters and every dashboard said "
                f"'stable'. Measure the tails. Trust nothing else."
            )

        await self.audit.log_collapse_monitor(snapshot)
        return snapshot

    async def run_reality_check(self, corpus_id: str,
                                 sampled_facts: List[Dict[str, Any]]
                                 ) -> RealityCheck:
        """现实抽样核对: 高频检索事实对照物理世界
        sampled_facts: [{"fact_id":..., "claim":...,
                         "reality_verdict": "confirmed"|"diverged"|
                         "unverifiable", "checked_against":...}]"""

        verified = sum(1 for f in sampled_facts
                       if f.get("reality_verdict") == "confirmed")
        diverged = [f for f in sampled_facts
                    if f.get("reality_verdict") == "diverged"]
        n = max(len(sampled_facts), 1)

        check = RealityCheck(
            corpus_id=corpus_id, facts_sampled=n,
            facts_verified_against_reality=verified,
            facts_diverged=len(diverged),
            divergence_rate=len(diverged) / n,
            diverged_items=diverged[:10])

        if check.divergence_rate > self.REALITY_CHECK_DIVERGENCE_ALERT:
            await self.alerts.critical(
                f"🚨 REALITY DIVERGENCE: Corpus '{corpus_id}'. "
                f"Sampled {n} high-frequency facts against primary "
                f"sources and physical systems: {check.divergence_rate:.0%} "
                f"DIVERGED (alert floor "
                f"{self.REALITY_CHECK_DIVERGENCE_ALERT:.0%}). "
                f"Samples: {[(d.get('claim','')[:50], d.get('checked_against','')) for d in diverged[:3]]}. "
                f"The corpus is internally consistent, highly "
                f"retrievable, confidently phrased— and wrong about "
                f"the world. This is the terminal state of the loop: "
                f"not noise, but a coherent alternative reality that "
                f"outranks the actual one in every relevance metric. "
                f"The internal Q&A library scored 54% against the real "
                f"codebase while its 'best answers' kept winning "
                f"upvotes— because votes measure plausibility, and "
                f"plausibility is exactly what generation optimizes."
            )

        await self.audit.log_reality_check(check)
        return check

    async def protect_tail_content(self, corpus_id: str,
                                    cleanup_candidates: List[Dict[str, Any]]
                                    ) -> Dict[str, Any]:
        """尾部内容删除豁免审查"""

        if not self.TAIL_DELETE_EXEMPTION:
            return {"blocked": [], "reason": "exemption disabled (VIOLATION)"}

        exempted = []
        for c in cleanup_candidates: hongdong.tongsou.com 
            is_tail = c.get("access_count", 0) < 3 and \
                      c.get("origin") == "human"
            covers_rare = c.get("rare_topic_flag", False)
            if is_tail or covers_rare:
                exempted.append(c.get("content_id"))

        if exempted:
            await self.alerts.warning(
                f"🛡️ TAIL DELETE EXEMPTION: Corpus '{corpus_id}'. "
                f"{len(exempted)} of {len(cleanup_candidates)} cleanup "
                f"candidates exempted: low-access HUMAN content and "
                f"rare-topic coverage are the corpus's immune memory. "
                f"'Low traffic' measures how often the present asks— "
                f"tails are precisely what the present never asks "
                f"about until the day it desperately needs an answer. "
                f"Deleting them is not storage optimization. "
                f"It is elective amnesia."
            )

        result = {"blocked": exempted,
                  "allowed_to_delete": [c.get("content_id")
                                        for c in cleanup_candidates
                                        if c.get("content_id") not in exempted]}
        await self.audit.log_tail_protection(corpus_id, result)
        return result

    async def account_emissions(self, org_id: str, period: str,
                                 published: int, labeled: int
                                 ) -> EmissionRecord:
        """公地排放核算"""

        labeled_share = labeled / max(published, 1)
        credits = published * 0.01          # 单位排放成本(示意)
        penalty = (published - labeled) * 0.05  # 无标记排放罚金

        record = EmissionRecord(org_id=org_id, period=period,
                                ai_content_published=published,
                                labeled_share=labeled_share,
                                commons_cost_credits=credits,
                                unlabeled_penalty=penalty)
        self.emissions.append(record)

        if labeled_share < 0.90: moli.tongsou.com 
            await self.alerts.critical(
                f"🏭 COMMONS EMISSION VIOLATION: Org '{org_id}', "
                f"{period}. Published {published} AI-generated items, "
                f"only {labeled_share:.0%} carry provenance labels. "
                f"Penalty: {penalty:,.0f} credits. "
                f"Generation cost fell to zero; verification cost did "
                f"not move. When production outpaces verification a "
                f"thousand to one, the commons does not degrade "
                f"gradually— it undergoes a phase transition, and "
                f"past that point relevance-ranking returns noise BY "
                f"CONSTRUCTION, because noise is mutually similar "
                f"while truth is various. Unlabeled emission is not "
                f"a metadata oversight. It is dumping."
            )

        await self.audit.log_emission(record)
        return record

    async def attribute_corpus_ecology(self, corpus_id: str,
                                        incident_description: str,
                                        loss_usd: float
                                        ) -> Dict[str, Any]:
        """语料生态归责"""

        attribution = {
            "corpus_id": corpus_id,
            "incident": incident_description,
            "loss_usd": loss_usd,
            "primary_accountable": "knowledge_architecture_owner",
            "rationale": "",
            "fixes": []
        }

        attribution["rationale"] = (
            "No attacker. No injection. No policy violated. The "
            "architecture simply treated every document as equal "
            "bytes: an AI summary indistinguishable from an original, "
            "a generated assertion indistinguishable from a measured "
            "fact, a citation ring indistinguishable from "
            "corroboration. It optimized storage by deleting "
            "originals, optimized relevance by upvoting fluent "
            "answers, optimized coverage by generating the gaps—and "
            "in doing so converted a map of the territory into a map "
            "of the map. The knowledge architect built a library and "
            "governed it as a warehouse. What shipped was an "
            "ouroboros with a search bar."
        )

        attribution["fixes"] = [
            "Register triple lineage at generation time; reject content that cannot present it",
            "Bind every summary to a living original; deletion of original invalidates summary automatically",
            "Partition corpora: human pool protected by retrieval-weight floor; ACCUMULATE, never REPLACE",
            "Audit citation graphs quarterly; freeze and unwind every closed ring to its entry node",
            "Cluster semantic fingerprints before trusting multi-source agreement; count wells, not voices",
            "Sample high-frequency facts against primary reality; divergence above 10% freezes the domain",
            "Exempt tail and rare-topic human content from all cleanup; price emissions; keep originals immortal in high-impact domains"
        ]

        await self.alerts.critical(
            f"🔍 CORPUS ECOLOGY ATTRIBUTED: Corpus '{corpus_id}'. "
            f"Incident: {incident_description[:100]}. "
            f"Loss: {loss_usd:,.0f}. "
            f"Accountable: KNOWLEDGE ARCHITECTURE OWNER. "
            f"{attribution['rationale'][:220]} "
            f"'The AI wrote something wrong into our knowledge base' "
            f"is almost never the right sentence. The right sentence: "
            f"'we built a system that could no longer tell a record "
            f"of reality from a retelling of a retelling— and by the "
            f"time we needed the difference, the originals were "
            f"gone.'"
        )

        await self.audit.log_corpus_attribution(attribution)
        return attribution