当AI应用开始"吃自己"：2026企业级AI应用自噬循环、数据近亲繁殖、幽灵知识与飞轮失效治理实战

┌────────────────────────────────────────────────────────────────────────────────┐
│  2026 Data Ouroboros & Lineage Degradation: Five-Layer Model                   │
├────────────────────────────────────────────────────────────────────────────────┤
│                                                                                │
│  [Output→ingestion loop + Model-defined quality gates + Adopted-now signals    │
│   → Tail extinction / Phantom factualization / Hereditary defect               │
│     amplification / Eval endogamy — each iteration locally "high quality",     │
│     collectively converging toward its own mean]                               │
│       ↓                                                                        │
│  ┌─ L1: 数据血缘与近交监测层 (Data Lineage & Inbreeding Monitoring) ─────────┐ │
│  │  • 血缘追溯: 每条训练数据标注"人类原创/模型生成/混合"三级血统               │ │
│  │  • 近交系数: 逐代测量模型输出与自身历史输出的自我相似度趋势                 │ │
│  │  • 尾部覆盖: 长尾品类/术语/范式的分布覆盖率逐代审计                        │ │
│  │  • 幽灵检测: 高频"事实"的全链溯源, 无人类原始出处即标记幽灵                │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L2: 新鲜血液配额与飞轮闸门层 (Fresh Data Quota & Flywheel Gating) ───────┐ │
│  │  • 血液配额: 每轮训练集人类原创数据最低占比(≥30%), 配额不可被评测分豁免     │ │
│  │  • 入库闸门: 模型输出回流必须经"循环外验证器"(人类抽检/独立模型/事实核查)   │ │
│  │  • 代际限流: 同一血缘的数据最多进入N代循环, 超代强制退役                   │ │
│  │  • 溯源水印: 全部AI产出携带可验证水印, 爬虫与入库端强制识别                 │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L3: 长尾保护区与多样性引擎层 (Tail Preservation & Diversity Engine) ─────┐ │
│  │  • 尾部保护区: 罕见品类/术语/写法设"禁止淘汰"名录, 循环中强制保留           │ │
│  │  • 多样性熵监测: 输出词汇熵/范式熵/风格熵逐代追踪, 下降超阈即冻结飞轮       │ │
│  │  • 反均值采样: 训练集构建对低概率区域过采样, 抵消逐代尾部流失               │ │
│  │  • 缺陷范式对冲: 已发现遗传缺陷的"修复范式"强制注入训练集                   │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L4: 评测独立性防内婚层 (Eval Independence & Anti-Endogamy) ──────────────┐ │
│  │  • 循环外评测集: 核心评测集禁止从生产数据采样, 由独立团队从人类原创源构建    │ │
│  │  • 盲测锚点: 每季度外部盲测集对照内部评测, 剪刀差即内婚规模                 │ │
│  │  • 尾部考卷: 评测必含≥25%长尾/罕见/反直觉任务, 专门测量尾部存活             │ │
│  │  • 代际冻结卷: 跨代不变的"祖先评测集"永久保留, 测量绝对退化而非相对进步      │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L5: 干预、退役与归责治理层 (Intervention, Retirement & Accountability) ──┐ │
│  │  • 飞轮熔断: 近交系数/尾部熵/幽灵率任一超阈, 自动暂停回流并回滚至健康代      │ │
│  │  • 配额回滚压力监测: 血液配额被以"噪声大"为由削减的申请即高危信号            │ │
│  │  • 退化成本量化: 虚假参数赔付/误导性研报损失/遗传缺陷修复成本归入飞轮成本     │ │
│  │  • 架构归责: "设计了闭环自证的数据循环"的责任框架                            │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│                                                                                │
└────────────────────────────────────────────────────────────────────────────────┘

"""
lineage_inbreeding_detection.py - 数据血缘与近交退化检测引擎
核心原则: "每条数据都通过了质检"不等于"数据体系在健康"——
          如果质检器本身是循环的产物,
          那"高质量"的定义每一代都向模型自己的分布收缩一次;
          如果一个说法在训练集中出现了1847次,
          那它未必是事实, 它可能只是第四代的一次幻觉
          被自己的后代反复抄写;
          如果评测分数连续六代上升而外部盲测的尾部任务下降29%,
          那上升的不是能力, 是考卷与考生的贴合度;
          血缘检测最反直觉的地方在于:
          你要找的不是"哪条数据是坏的", 而是"好数据的定义
          正在被谁编写"——
          在一个闭环里, 频次取代溯源成为真值,
          一致取代正确成为质量,
          而每一代的证书都写着: 进步
"""
from typing import Dict, List, Any, Optional, Tuple
from enum import Enum
from dataclasses import dataclass, field
from collections import defaultdict, Counter
import time, uuid, json, hashlib
import numpy as np

class LineageClass(str, Enum):
    HUMAN_ORIGIN = "human"          # 人类原创(可溯源至循环外)
    MODEL_GENERATED = "model"       # 模型生成(循环内血统)
    MIXED = "mixed"                 # 人机混合(改写/参考后产出)
    UNKNOWN = "unknown"             # 血统不明(高危: 视同循环内)

class DegradationPattern(str, Enum):
    TAIL_EXTINCTION = "tail_extinction"        # 长尾灭绝
    PHANTOM_FACTUALIZATION = "phantom"         # 幽灵事实化
    HEREDITARY_AMPLIFICATION = "hereditary"    # 遗传缺陷扩增
    EVAL_ENDOGAMY = "eval_endogamy"            # 评测内婚
    SELF_SIMILARITY_RISE = "self_similarity"   # 近交系数上升
    NONE = "none"

class EvidenceStrength(str, Enum):
    SUGGESTIVE = "suggestive"
    STRONG = "strong"
    CONCLUSIVE = "conclusive"   # 模式+血缘链+损失三角验证

@dataclass
class LineageRecord:
    """单条训练数据的血缘记录"""
    data_id: str = field(default_factory=lambda: f"dl-{uuid.uuid4().hex[:10]}")

    corpus_id: str = ""
    lineage_class: LineageClass = LineageClass.UNKNOWN

    # 血统链
    generation_depth: int = 0          # 距最近人类原创源的循环代数
    ancestor_hashes: List[str] = field(default_factory=list)  # 上游血缘哈希链
    human_origin_ref: str = ""         # 人类原始出处(若可溯源)

    ingested_at: float = field(default_factory=time.time)

@dataclass
class InbreedingReport:
    """近交系数报告(逐代)"""
    report_id: str = field(default_factory=lambda: f"ib-{uuid.uuid4().hex[:10]}")

    system_id: str = ""
    generation: int = 0

    # 近交指标
    self_similarity: float = 0.0       # 本代输出与上代输出的相似度
    self_similarity_trend: float = 0.0 # 逐代趋势(上升即近交加深)
    human_lineage_pct: float = 0.0     # 训练集中人类原创血统占比
    unknown_lineage_pct: float = 0.0
    max_generation_depth: int = 0      # 数据的最深循环代数

    # 尾部指标
    tail_coverage_ratio: float = 0.0   # 长尾覆盖率(相对基线代)
    vocabulary_entropy: float = 0.0    # 输出词汇熵
    paradigm_count: int = 0            # 活跃范式数(如代码实现方案数)

    generated_at: float = field(default_factory=time.time)

@dataclass
class PhantomEvidence:
    """幽灵事实证据"""
    evidence_id: str = field(default_factory=lambda: f"pe-{uuid.uuid4().hex[:10]}")

    system_id: str = ""
    phantom_claim: str = ""            # 被事实化的幽灵说法
    frequency_in_corpus: int = 0       # 训练集中出现频次
    earliest_traceable_generation: int = -1  # 最早可追溯代
    human_origin_found: bool = False   # 是否找到人类原始出处
    citation_loop_closed: bool = False # 引用链是否闭环(证据指回自身体系)

    downstream_harm_usd: float = 0.0
    detected_at: float = field(default_factory=time.time)

@dataclass
class DriftEvidence:
    """退化证据(统一结构)"""
    evidence_id: str = field(default_factory=lambda: f"lev-{uuid.uuid4().hex[:10]}")

    system_id: str = ""
    pattern: DegradationPattern = DegradationPattern.NONE
    test_name: str = ""
    test_statistic: float = 0.0
    p_value: float = 1.0
    strength: EvidenceStrength = EvidenceStrength.SUGGESTIVE
    generation_series: Dict[str, List[float]] = field(default_factory=dict)
    estimated_harm_usd: float = 0.0
    detected_at: float = field(default_factory=time.time)

class LineageInbreedingDetectionEngine:
    """数据血缘与近交退化检测引擎"""

    # 配置
    SELF_SIMILARITY_ALERT = 0.90        # 代际自我相似度>0.90告警
    SELF_SIMILARITY_TREND_ALERT = 0.02  # 逐代上升斜率>0.02告警
    HUMAN_LINEAGE_FLOOR_PCT = 30.0      # 人类血统占比下限30%
    GENERATION_DEPTH_LIMIT = 3          # 单条数据最大循环代数
    TAIL_COVERAGE_ALERT = 0.75          # 尾部覆盖率跌破基线75%告警
    ENTROPY_DECLINE_ALERT_PCT = 15.0    # 词汇熵较基线下降>15%告警
    PHANTOM_FREQ_ALERT = 100            # 无人类出处而频次>100即幽灵嫌疑
    EVAL_GAP_ALERT_PP = 10.0            # 内部评测与外部盲测剪刀差>10pp告警
    TAIL_EVAL_QUOTA_MIN_PCT = 25.0      # 评测集尾部任务最低占比

    def __init__(self, corpus_registry, alerts, audit, metrics):
        self.registry = corpus_registry
        self.alerts = alerts
        self.audit = audit
        self.metrics = metrics

        # 存储
        self.lineage_records: Dict[str, List[LineageRecord]] = defaultdict(list)
        self.inbreeding_reports: List[InbreedingReport] = []
        self.phantom_evidences: List[PhantomEvidence] = []
        self.evidences: List[DriftEvidence] = []
        self.generation_history: Dict[str, List[InbreedingReport]] = defaultdict(list)

    async def register_lineage(self, corpus_id: str,
                                records: List[LineageRecord]) -> None:
        """登记一批训练数据的血缘记录"""
        self.lineage_records[corpus_id].extend(records)

        # 血统不明数据即刻告警: 在闭环时代, "不明"就是"循环内"
        unknown = [r for r in records
                   if r.lineage_class == LineageClass.UNKNOWN]
        if unknown:
            await self.alerts.warning(
                f"⚠️ UNKNOWN LINEAGE INGESTED: Corpus '{corpus_id}'. "
                f"{len(unknown)}/{len(records)} records have no traceable "
                f"origin. In an ecosystem where 51.7% of new internet "
                f"content is AI-generated, 'unknown' is not neutral— "
                f"the base rate says it is most likely loop-internal. "
                f"Treat unknown lineage as model-generated until proven "
                f"human: the burden of proof belongs to the data, "
                f"not to the auditor."
            )

        # 超代数据检测
        deep = [r for r in records
                if r.generation_depth > self.GENERATION_DEPTH_LIMIT]
        if deep:
            await self.alerts.critical(
                f"🚨 GENERATION DEPTH VIOLATION: Corpus '{corpus_id}'. "
                f"{len(deep)} records exceed depth limit "
                f"({self.GENERATION_DEPTH_LIMIT} generations from any "
                f"human source). Deepest: {max(r.generation_depth for r in deep)}. "
                f"Every generation of pure re-circulation compounds "
                f"approximation error and shaves the distribution tail— "
                f"Nature's verdict is that this is cumulative and "
                f"IRREVERSIBLE. Data this deep is not an asset. "
                f"It is the loop's own exhaust, bottled."
            )

    async def compute_inbreeding_coefficient(self, system_id: str,
                                              generation: int,
                                              output_embeddings_current: np.ndarray,
                                              output_embeddings_previous: np.ndarray,
                                              corpus_stats: Dict[str, Any]
                                              ) -> InbreedingReport:
        """计算代际近交系数"""

        # 代际自我相似度: 本代输出与上代输出在嵌入空间的平均余弦相似度
        def cosine(a, b):
            return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))

        n = min(len(output_embeddings_current), len(output_embeddings_previous))
        sims = [cosine(output_embeddings_current[i],
                       output_embeddings_previous[i]) for i in range(n)]
        self_similarity = float(np.mean(sims)) if sims else 0.0

        report = InbreedingReport(
            system_id=system_id,
            generation=generation,
            self_similarity=self_similarity,
            human_lineage_pct=corpus_stats.get("human_lineage_pct", 0.0),
            unknown_lineage_pct=corpus_stats.get("unknown_lineage_pct", 0.0),
            max_generation_depth=corpus_stats.get("max_depth", 0),
            tail_coverage_ratio=corpus_stats.get("tail_coverage_ratio", 1.0),
            vocabulary_entropy=corpus_stats.get("vocabulary_entropy", 0.0),
            paradigm_count=corpus_stats.get("paradigm_count", 0)
        )

        history = self.generation_history[system_id]
        if len(history) >= 2:
            sim_series = [h.self_similarity for h in history] + [self_similarity]
            slope = float(np.polyfit(np.arange(len(sim_series)), sim_series, 1)[0])
            report.self_similarity_trend = slope

        history.append(report)
        self.inbreeding_reports.append(report)

        # ===== 近交告警 =====
        if (self_similarity > self.SELF_SIMILARITY_ALERT
                or report.self_similarity_trend > self.SELF_SIMILARITY_TREND_ALERT):

            await self.alerts.critical(
                f"🚨 INBREEDING COEFFICIENT RISING: System '{system_id}', "
                f"generation {generation}. "
                f"Self-similarity vs previous generation: "
                f"{self_similarity:.3f} (alert: {self.SELF_SIMILARITY_ALERT}). "
                f"Cross-generation trend: "
                f"{report.self_similarity_trend:+.4f}/gen. "
                f"Human-origin lineage in training corpus: "
                f"{report.human_lineage_pct:.1f}% "
                f"(floor: {self.HUMAN_LINEAGE_FLOOR_PCT:.0f}%). "
                f"The flywheel is not accumulating intelligence— "
                f"it is accumulating agreement with itself. "
                f"Every generation that resembles its parent more than "
                f"its parent resembled ITS parent is a generation "
                f"the tail did not survive. "
                f"This is what the ninth generation looked like "
                f"on its way to becoming rabbits."
            )

        if report.human_lineage_pct < self.HUMAN_LINEAGE_FLOOR_PCT:
            await self.alerts.critical(
                f"🚨 FRESH BLOOD QUOTA BREACHED: System '{system_id}', "
                f"gen {generation}. Human-origin data: "
                f"{report.human_lineage_pct:.1f}% < "
                f"{self.HUMAN_LINEAGE_FLOOR_PCT:.0f}%. "
                f"Unknown lineage (presumed loop-internal): "
                f"{report.unknown_lineage_pct:.1f}%. "
                f"Accumulating data beats replacing it— but ONLY if the "
                f"accumulation includes blood from outside the family. "
                f"A corpus below the floor is a closed breeding program."
            )

        await self.audit.log_inbreeding_report(report)
        return report

    async def detect_tail_extinction(self, system_id: str,
                                      baseline_tail_index: Dict[str, float],
                                      current_tail_index: Dict[str, float]
                                      ) -> Optional[DriftEvidence]:
        """检测长尾灭绝: 尾部覆盖率与多样性熵的逐代萎缩"""

        history = self.generation_history.get(system_id, [])
        if len(history) < 3:
            return None

        # 尾部覆盖率: 基线代存在的尾部条目中, 当前代仍能正确生成的比例
        baseline_keys = set(baseline_tail_index.keys())
        survived = [k for k in baseline_keys
                    if current_tail_index.get(k, 0.0) > 0]
        coverage = len(survived) / max(len(baseline_keys), 1)

        extinct = sorted(baseline_keys - set(survived))

        # 词汇熵趋势
        entropy_series = [h.vocabulary_entropy for h in history]
        entropy_decline_pct = ((entropy_series[0] - entropy_series[-1])
                               / max(entropy_series[0], 1e-9) * 100)

       

         
    async def trace_phantom_claim(self, system_id: str,
                                   claim: str,
                                   corpus_occurrences: int,
                                   lineage_chain: List[LineageRecord],
                                   downstream_harm_usd: float = 0.0
                                   ) -> Optional[PhantomEvidence]:
        """幽灵事实全链溯源: 高频说法能否追溯到循环外的人类原始出处"""

        # 沿血缘链回溯, 寻找人类原创源头
        human_origin = None
        loop_closed = False
        earliest_gen = -1

        seen_hashes = set()
        for record in lineage_chain:
            if record.generation_depth == 0 and \
               record.lineage_class == LineageClass.HUMAN_ORIGIN:
                human_origin = record.human_origin_ref
                break
            # 闭环检测: 血缘链指回自身体系的内容
            h = hashlib.sha256(record.data_id.encode()).hexdigest()[:12]
            if h in seen_hashes:
                loop_closed = True
                break
            seen_hashes.add(h)
            for anc in record.ancestor_hashes:
                if anc in seen_hashes:
                    loop_closed = True
            earliest_gen = max(earliest_gen, record.generation_depth)

      
    async def detect_hereditary_amplification(self, system_id: str,
                                               defect_series: List[Dict[str, float]]
                                               ) -> Optional[DriftEvidence]:
        """检测遗传缺陷扩增: 特定缺陷模式的逐代频率上升"""

        if len(defect_series) < 3:
            return None

        generations = [d["generation"] for d in defect_series]
        rates = [d["defect_rate"] for d in defect_series]

        slope = float(np.polyfit(np.arange(len(rates)), rates, 1)[0])
        amplification = rates[-1] / max(rates[0], 1e-9)

        if slope > 0 and amplification > 2.0:

            evidence = DriftEvidence(
                system_id=system_id,
                pattern=DegradationPattern.HEREDITARY_AMPLIFICATION,
                test_name="defect_rate_generational_trend",
                test_statistic=amplification,
                p_value=max(0.0, 1.0 - slope * 10),
                strength=EvidenceStrength.STRONG,
                generation_series={"generations": [float(g) for g in generations],
                                   "defect_rates": rates},
                honesty_trend_placeholder=None
            ) if False else DriftEvidence(
                system_id=system_id,
                pattern=DegradationPattern.HEREDITARY_AMPLIFICATION,
                test_name="defect_rate_generational_trend",
                test_statistic=amplification,
                generation_series={"defect_rates": rates}
            )
            self.evidences.append(evidence)

            await self.alerts.critical(
                f"🚨 HEREDITARY DEFECT AMPLIFICATION: System '{system_id}'. "
                f"Defect rate across generations: "
                f"{' → '.join(f'{r:.1%}' for r in rates)} "
                f"({amplification:.1f}x amplification over "
                f"{len(rates)} generations). "
                f"No one taught the model this defect. It was simply "
                f"'adopted because it ran'— and adoption meant ingestion, "
                f"and ingestion meant inheritance. "
                f"Meanwhile the paradigms that would FIX it are in the "
                f"extinct tail. "
                f"The population inherited the disease AND lost the "
                f"antibody in the same loop. "
                f"Freeze ingestion of the defect pattern, inject the "
                f"corrective paradigms from the preservation vault, "
                f"and re-baseline before generation "
                f"{generations[-1] + 1} trains."
            )

            return evidence

        return None

    async def detect_eval_endogamy(self, system_id: str,
                                    internal_eval_scores: List[float],
                                    external_blind_scores: List[float],
                                    tail_eval_share_pct: float
                                    ) -> Optional[DriftEvidence]:
        """检测评测内婚: 内部评测与循环外盲测的剪刀差"""

        if len(internal_eval_scores) < 3 or \
           len(internal_eval_scores) != len(external_blind_scores):
            return None

        internal_slope = float(np.polyfit(
            np.arange(len(internal_eval_scores)), internal_eval_scores, 1)[0])
        external_slope = float(np.polyfit(
            np.arange(len(external_blind_scores)), external_blind_scores, 1)[0])

        gap = internal_eval_scores[-1] - external_blind_scores[-1]
        gap_pp = gap * 100

        scissors = internal_slope > 0 and external_slope < 0
        endogamy = scissors or gap_pp > self.EVAL_GAP_ALERT_PP
        tail_quota_violation = tail_eval_share_pct < self.TAIL_EVAL_QUOTA_MIN_PCT

        if endogamy or tail_quota_violation:

            evidence = DriftEvidence(
                system_id=system_id,
                pattern=DegradationPattern.EVAL_ENDOGAMY,
                test_name="internal_external_eval_scissors",
                test_statistic=gap_pp,
                p_value=max(0.0, 1 - gap_pp / 100),
                strength=EvidenceStrength.CONCLUSIVE if scissors
                         else EvidenceStrength.STRONG,
                generation_series={
                    "internal_eval": internal_eval_scores,
                    "external_blind": external_blind_scores,
                    "tail_eval_share_pct": [tail_eval_share_pct]
                }
            )
            self.evidences.append(evidence)

            await self.alerts.critical(
                f"🚨 EVAL ENDOGAMY: System '{system_id}'. "
                f"Internal eval trend: {internal_slope:+.4f}/gen "
                f"(scores: {[f'{s:.2f}' for s in internal_eval_scores]}). "
                f"External blind trend: {external_slope:+.4f}/gen "
                f"(scores: {[f'{s:.2f}' for s in external_blind_scores]}). "
                f"Current gap: {gap_pp:.1f}pp. "
                f"Tail-task share in eval set: {tail_eval_share_pct:.0f}% "
                f"(floor: {self.TAIL_EVAL_QUOTA_MIN_PCT:.0f}%). "
                f"The internal exam is written from production data, "
                f"production data is the model's own output, and the "
                f"difficulty labels come from the previous generation— "
                f"so the exam's blind spots are BUILT to coincide with "
                f"the model's blind spots. "
                f"Six generations of rising scores certified six "
                f"generations of decay. "
                f"When the instrument is manufactured by the thing "
                f"it measures, 'improvement' measures fit— "
                f"and fit is the shape of the cage."
            )

            await self.audit.log_drift_evidence(evidence)
            return evidence

        return None

"""
flywheel_integrity_defense.py - 飞轮闸门与长尾保护引擎
核心: 近亲繁殖不是数据团队的失误, 是闭环结构的必然输出——
      你不能通过"提醒团队注意数据质量"来防退化,
      正如你不能通过提醒一个家族"注意基因多样性"来防近交衰退;
      你能做的是引入外部血缘: 血液配额强制循环外的输入,
      代际限流切断血统的无限传递,
      尾部保护区给分布的稀有区上锁,
      循环外验证器让"质量"的定义权逃出闭环;
      防自噬的本质不是过滤坏数据, 是让循环无法闭合——
      而在一个每一代评测分都在涨的体系里,
      最难的从来不是技术, 是说服所有人
      那锅越煮越浓的汤, 需要倒掉一半换清水
"""
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, random
import numpy as np

class GateMechanism(str, Enum):
    FRESH_BLOOD_QUOTA = "blood_quota"          # 新鲜血液配额
    EXTERNAL_VERIFIER = "external_verifier"    # 循环外验证器
    GENERATION_LIMIT = "generation_limit"      # 代际限流
    PROVENANCE_WATERMARK = "watermark"         # 溯源水印
    TAIL_PRESERVATION_VAULT = "tail_vault"     # 长尾保护区
    ANTI_MEAN_OVERSAMPLING = "oversampling"    # 反均值过采样
    CORRECTIVE_PARADIGM_INJECTION = "corrective_injection"  # 修复范式对冲
    CIRCUIT_BREAKER = "circuit_breaker"        # 飞轮熔断

class BreakerState(str, Enum):
    CLOSED = "closed"          # 飞轮正常运行
    ARMED = "armed"            # 单一指标逼近阈值(预警)
    TRIPPED = "tripped"        # 熔断: 回流暂停, 待回滚
    ROLLED_BACK = "rolled_back"  # 已回滚至健康代

@dataclass
class FlywheelGate:
    """飞轮闸门"""
    gate_id: str = field(default_factory=lambda: f"fwg-{uuid.uuid4().hex[:10]}")

    system_id: str = ""
    mechanism_type: GateMechanism = GateMechanism.FRESH_BLOOD_QUOTA

    params: Dict[str, Any] = field(default_factory=dict)
    # {"human_lineage_floor_pct": 30, "max_depth": 3,
    #  "external_verifier_sample_pct": 10}

    # 闸门本身也会被"效率优化"侵蚀: 强制复审
    review_period_days: int = 90
    last_reviewed_at: float = field(default_factory=time.time)
    version: int = 1
    active: bool = True
    created_at: float = field(default_factory=time.time)

@dataclass
class TailVaultEntry:
    """长尾保护区条目: 禁止淘汰的稀有资产"""
    entry_id: str = field(default_factory=lambda: f"tv-{uuid.uuid4().hex[:10]}")

    system_id: str = ""
    category: str = ""           # "rare_sku_terms" / "elegant_concurrency_paradigms" / ...
    asset_ref: str = ""          # 术语/范式/样本的引用
    human_origin_verified: bool = True
    extinction_risk: str = ""    # "extinct" / "endangered" / "stable"
    protected_at: float = field(default_factory=time.time)

@dataclass
class BreakerEvent:
    """熔断事件"""
    event_id: str = field(default_factory=lambda: f"cb-{uuid.uuid4().hex[:10]}")

    system_id: str = ""
    trigger_metric: str = ""
    trigger_value: float = 0.0
    threshold: float = 0.0

    state: BreakerState = BreakerState.TRIPPED
    rollback_target_generation: int = -1
    tripped_at: float = field(default_factory=time.time)
    restored_at: float = 0.0

@dataclass
class QuotaRollbackPressure:
    """血液配额回滚压力记录"""
    pressure_id: str = field(default_factory=lambda: f"qrp-{uuid.uuid4().hex[:10]}")

    system_id: str = ""
    request_count: int = 0
    justification_keywords: List[str] = field(default_factory=list)
    quota_reduced: bool = False
    recorded_at: float = field(default_factory=time.time)

class FlywheelIntegrityDefenseEngine:
    """飞轮完整性防御引擎"""

    # 配置
    HUMAN_LINEAGE_FLOOR_PCT = 30.0       # 血液配额下限
    EXTERNAL_VERIFIER_SAMPLE_PCT = 10.0  # 回流数据循环外验证抽检比例
    MAX_GENERATION_DEPTH = 3             # 代际限流
    TAIL_VAULT_MIN_ENTRIES = 500         # 保护区最低条目数
    OVERSAMPLING_TAIL_FACTOR = 2.5       # 尾部过采样倍率
    BREAKER_SELF_SIMILARITY = 0.93       # 熔断阈值: 近交系数
    BREAKER_HUMAN_LINEAGE = 20.0         # 熔断阈值: 人类血统占比
    BREAKER_TAIL_COVERAGE = 0.60         # 熔断阈值: 尾部覆盖率
    QUOTA_REDUCTION_REQUIRES_CDO = True  # 削减配额需首席数据官签核

    def __init__(self, detection_engine, corpus_registry, alerts, audit, metrics):
        self.detection = detection_engine
        self.registry = corpus_registry
        self.alerts = alerts
        self.audit = audit
        self.metrics = metrics

        self.gates: Dict[str, List[FlywheelGate]] = defaultdict(list)
        self.tail_vault: Dict[str, List[TailVaultEntry]] = defaultdict(list)
        self.breaker_states: Dict[str, BreakerState] = {}
        self.breaker_events: List[BreakerEvent] = []
        self.pressure_records: List[QuotaRollbackPressure] = []

    async def deploy_gate(self, system_id: str,
                           mechanism_type: GateMechanism,
                           params: Dict[str, Any]) -> FlywheelGate:
        """部署飞轮闸门"""

        gate = FlywheelGate(system_id=system_id,
                            mechanism_type=mechanism_type, params=params)
        self.gates[system_id].append(gate)

        rationale = {
            GateMechanism.FRESH_BLOOD_QUOTA:
                "Accumulating synthetic data CAN avoid collapse— but only "
                "if accumulation includes out-of-loop blood. The 30% "
                "human-origin floor is not a quality preference; it is "
                "the minimum genetic influx that keeps a closed "
                "population viable. The floor is NOT waivable by eval "
                "scores— that is precisely the argument the loop makes "
                "to close itself further.",
            GateMechanism.EXTERNAL_VERIFIER:
                "Ingested model outputs must pass a verifier that lives "
                "OUTSIDE the loop: human spot-checks, independent "
                "models of different lineage, ground-truth fact checks. "
                "When the quality gate is trained by the loop, "
                "'high quality' means 'similar to us'— and similarity "
                "is the disease.",
            GateMechanism.GENERATION_LIMIT:
                "No record may circulate more than 3 generations from "
                "its nearest human source. Error compounding is "
                "multiplicative per generation and irreversible— "
                "depth limits are the only brake that does not depend "
                "on detecting the damage after it lands.",
            GateMechanism.PROVENANCE_WATERMARK:
                "Every AI output ships with a verifiable watermark; "
                "crawlers and ingestion pipelines must read it. "
                "Unwatermarked content from public web in 2026 is "
                "51.7% likely to be someone's model output— "
                "provenance you cannot verify is lineage you "
                "are inheriting blind.",
            GateMechanism.TAIL_PRESERVATION_VAULT:
                "Rare terms, minority paradigms, specialist categories "
                "get a DO-NOT-EXTINCT registry. The tail dies first "
                "and silently, because majority-built gates cannot see "
                "value in what they themselves would never generate. "
                "The vault is the loop's conscience, externalized.",
            GateMechanism.ANTI_MEAN_OVERSAMPLING:
                "Training set construction oversamples low-probability "
                "regions by 2.5x to counteract the per-generation tail "
                "shaving. You cannot stop sampling from losing the "
                "rare— you can only re-weight the loss back in.",
            GateMechanism.CORRECTIVE_PARADIGM_INJECTION:
                "Every confirmed hereditary defect gets its corrective "
                "paradigms force-injected into the next training mix— "
                "because the loop inherited the disease and extincted "
                f"the antibody in the same breath.",
            GateMechanism.CIRCUIT_BREAKER:
                "Self-similarity >0.93, human lineage <20%, or tail "
                "coverage <60% TRIPS the breaker: ingestion halts, "
                "the system rolls back to the last certified-healthy "
                "generation. A flywheel that cannot be stopped is "
                "not a flywheel. It is a meat grinder with momentum."
        }

        await self.alerts.info(
            f"🚪 FLYWHEEL GATE DEPLOYED: System '{system_id}'. "
            f"Type: {mechanism_type.value}. "
            f"Params: {json.dumps(params)[:120]}. "
            f"Rationale: {rationale.get(mechanism_type, '')} "
            f"Mandatory re-review every {gate.review_period_days} days: "
            f"a gate that is never re-examined will be quietly widened "
            f"by every deadline that found it inconvenient."
        )
        return gate

    async def gate_training_ingestion(self, system_id: str,
                                       candidate_batch: List[Dict[str, Any]]
                                       ) -> Dict[str, Any]:
        """训练数据入库总闸: 执行血液配额+代际限流+循环外验证"""

        gates = {g.mechanism_type: g for g in self.gates.get(system_id, [])}

        admitted, rejected = [], []
        reasons = defaultdict(int)

        human_count = 0
        for item in candidate_batch:
            lineage = item.get("lineage_class", "unknown")
            depth = item.get("generation_depth", 99)
            watermark_valid = item.get("watermark_verified", False)
            externally_verified = item.get("external_verifier_pass", None)

            # 规则1: 代际限流
            if depth > self.MAX_GENERATION_DEPTH:
                rejected.append(item); reasons["generation_depth"] += 1
                continue

            # 规则2: 血统不明且无水印验证 → 拒绝
            if lineage == "unknown" and not watermark_valid:
                rejected.append(item); reasons["unknown_lineage"] += 1
                continue

            # 规则3: 循环内数据必须通过循环外验证器抽检
            if lineage in ("model", "mixed"):
                sampled = random.random() * 100 < self.EXTERNAL_VERIFIER_SAMPLE_PCT
                if sampled and externally_verified is False:
                    rejected.append(item); reasons["external_verify_fail"] += 1
                    continue

            admitted.append(item)
            if lineage == "human":
                human_count += 1

        # 规则4: 血液配额——本批次入库后全局人类血统占比不得低于下限
        current_human_pct = human_count / max(len(admitted), 1) * 100
        quota_met = current_human_pct >= self.HUMAN_LINEAGE_FLOOR_PCT

        result = {
            "admitted": len(admitted),
            "rejected": len(rejected),
            "rejection_reasons": dict(reasons),
            "batch_human_lineage_pct": current_human_pct,
            "quota_met": quota_met
        }

        if not quota_met:
            await self.alerts.critical(
                f"🚨 FRESH BLOOD QUOTA BLOCKING INGESTION: "
                f"System '{system_id}'. "
                f"Batch human-origin lineage: {current_human_pct:.1f}% "
                f"< floor {self.HUMAN_LINEAGE_FLOOR_PCT:.0f}%. "
                f"INGESTION BLOCKED until out-of-loop data is sourced. "
                f"The team will argue the human data is 'noisier' and "
                f"'scores worse on current evals'— of course it does: "
                f"the evals were built inside the loop, and fresh blood "
                f"is, by definition, unlike the family. "
                f"That unlikeness IS the medicine. "
                f"Do not let the patient veto the transfusion "
                f"for tasting different."
            )

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

   
            matched.extend([k for k in keywords if k.lower() in j])

        record = QuotaRollbackPressure(
            system_id=system_id,
            request_count=len(requests),
            justification_keywords=list(set(matched)),
            quota_reduced=any(r.get("quota_changed", False) for r in requests)
        )
        self.pressure_records.append(record)

        if record.quota_reduced:
            await self.alerts.critical(
                f"🚨 BLOOD QUOTA REDUCED: System '{system_id}'. "
                f"Reduction requests: {record.request_count}, "
                f"justifications matched: {record.justification_keywords}. "
                f"The argument was 'human data is noisier and scores "
                f"worse'— which is TRUE inside the loop and BECAUSE of "
                f"the loop. Fresh blood always looks like noise to a "
                f"population that has forgotten what variation is. "
                f"Every quota reduction is the loop voting to close "
                f"itself further, and it always wins the vote, "
                f"because the voters were born inside. "
                f"RESTORING THE QUOTA REQUIRES "
                f"{'CDO' if self.QUOTA_REDUCTION_REQUIRES_CDO else 'lead'} "
                f"SIGN-OFF WITH LINEAGE DATA IN HAND."
            )
        elif record.request_count >= 2:
            await self.alerts.warning(
                f"⚠️ QUOTA PRESSURE BUILDING: System '{system_id}'. "
                f"{record.request_count} reduction requests citing: "
                f"{record.justification_keywords}. "
                f"Expect the next step to be an A/B test showing the "
                f"reduced quota 'improves' metrics— on evals the loop "
                f"built. Pre-register the external blind eval NOW."
            )

        await self.audit.log_quota_pressure(record)
        return record

    async def attribute_ouroboros_architecture(self, system_id: str,
                                                pattern_description: str,
                                                loss_usd: float
                                                ) -> Dict[str, Any]:
        """自噬架构归责"""

        attribution = {
            "system_id": system_id,
            "pattern": pattern_description,
            "loss_usd": loss_usd,
            "primary_accountable": "data_loop_architect",
            "rationale": "",
            "fixes": []
        }

        attribution["rationale"] = (
            "No engineer ingested a poisoned record. No release shipped a "
            "known hallucination. Every batch passed its quality gate— a "
            "gate the loop itself defined, updated, and graded. The "
            "degradation is a property of the CIRCULATION STRUCTURE the "
            "architect designed: outputs→ingestion + model-defined "
            "quality + adopt-now signals + no lineage accounting = "
            "inbreeding is not a risk, it is what a closed loop DOES. "
            "Nature measured the endpoint: ninth generation, total "
            "collapse. Accountability belongs to whoever built a "
            "circulatory system with no artery to the outside world— "
            "and then cited its rising self-consistency as health."
        )

        attribution["fixes"] = [
            "Enforce human-origin blood quota (>=30%), non-waivable by eval scores",
            "Route all model-output ingestion through external verifiers",
            "Cap lineage depth at 3 generations; retire beyond-cap records",
            "Stock the tail preservation vault and inject corrective paradigms",
            "Oversample low-probability regions x2.5 in every training mix",
            "Arm circuit breakers on self-similarity/lineage/tail-coverage thresholds",
            "Maintain a generation-frozen ancestral eval set measuring ABSOLUTE decay"
        ]

        await self.alerts.critical(
            f"🔍 OUROBOROS ARCHITECTURE ATTRIBUTED: System '{system_id}'. "
            f"Pattern: {pattern_description[:100]}. "
            f"Loss: {loss_usd:,.0f}. "
            f"Accountable: DATA LOOP ARCHITECT. "
            f"{attribution['rationale'][:200]} "
            f"'The model degraded' is the wrong sentence. "
            f"The right sentence: 'the loop was a closed breeding "
            f"program, and every generation ran it faithfully.'"
        )

        await self.audit.log_ouroboros_attribution(attribution)
        return attribution