CIE-Unified

test_exactly_once.py

  1"""
  2Phase 2.1 Gate: PendingSignal exactly-once 回归测试
  3=====================================================
  4验证：
  51. emit() 产生的回灌不会被 ingest()+step() 双重应用
  62. commit_feedback() 不会和旧 _feedback_loop() 叠加
  73. PendingSignal 是唯一反馈入口
  84. _feedback_loop 方法不存在
  9"""
 10
 11import sys
 12import os
 13sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 14
 15from cie import CIERuntime
 16
 17
 18def test_no_feedback_loop_method():
 19    """_feedback_loop 方法已被彻底删除"""
 20    rt = CIERuntime(seed=42)
 21    assert not hasattr(rt, '_feedback_loop'), \
 22        "_feedback_loop method still exists — must be removed"
 23    assert not hasattr(rt, '_feedback_pending'), \
 24        "_feedback_pending attribute still exists — must be removed"
 25    print("  PASS: _feedback_loop and _feedback_pending fully removed")
 26
 27
 28def test_pending_signal_is_sole_entry():
 29    """PendingSignal 是唯一的反馈入口"""
 30    rt = CIERuntime(seed=42)
 31
 32    # 检查 runtime.py 源码中不包含 _feedback_loop
 33    import inspect
 34    source = inspect.getsource(rt.__class__)
 35    assert '_feedback_loop' not in source, \
 36        "_feedback_loop reference found in CIERuntime source"
 37    assert '_feedback_pending' not in source, \
 38        "_feedback_pending reference found in CIERuntime source"
 39    print("  PASS: PendingSignal is sole entry point (source verified)")
 40
 41
 42def test_emit_feedback_exactly_once():
 43    """emit() 的回灌只通过 PendingSignal 应用一次"""
 44    rt = CIERuntime(seed=42)
 45
 46    rt.ingest("测试")
 47    rt.step(n=3)
 48    out = rt.emit()
 49
 50    # emit 后应有一个 pending signal
 51    pending_count = len(rt.state.pending_signals)
 52    assert pending_count == 1, \
 53        f"Expected 1 pending signal after emit, got {pending_count}"
 54
 55    # 记录 emit 后的状态
 56    phi_after_emit = dict(rt.state.phi)
 57
 58    # ingest 新内容——不应该触发旧的 _feedback_loop
 59    rt.ingest("新输入")
 60
 61    # 现在应该有 2 个 pending signal（emit 的 + ingest 的）
 62    pending_count = len(rt.state.pending_signals)
 63    assert pending_count == 2, \
 64        f"Expected 2 pending signals after ingest, got {pending_count}"
 65
 66    # step 消费所有信号——每个信号只应用一次
 67    rt.step(n=1)
 68
 69    # pending 应该清空
 70    assert len(rt.state.pending_signals) == 0, \
 71        f"Pending signals not cleared after step: {len(rt.state.pending_signals)}"
 72
 73    print("  PASS: emit feedback applied exactly once via PendingSignal")
 74
 75
 76def test_commit_feedback_via_signal_only():
 77    """commit_feedback() 只创建信号，不直接修改状态"""
 78    rt = CIERuntime(seed=42)
 79
 80    rt.ingest("反馈测试")
 81    rt.step(n=3)
 82    rt.emit()
 83
 84    # 清空 pending signals
 85    rt.step(n=1)
 86    assert len(rt.state.pending_signals) == 0
 87
 88    # commit_feedback 应只创建信号
 89    phi_before = dict(rt.state.phi)
 90    rt.commit_feedback({'correct': ['反'], 'reward': 1.0})
 91
 92    # 状态不应立即改变（信号在队列里等待 step 消费）
 93    pending = len(rt.state.pending_signals)
 94    assert pending >= 1, \
 95        f"commit_feedback should enqueue signals, got {pending}"
 96
 97    # step 消费信号
 98    rt.step(n=1)
 99    assert len(rt.state.pending_signals) == 0
100
101    print("  PASS: commit_feedback works through signal queue only")
102
103
104def test_no_double_application():
105    """多轮 ingest→step→emit 不会导致反馈双重应用"""
106    rt = CIERuntime(seed=42)
107
108    mu_deltas = []
109    for round_i in range(5):
110        rt.ingest("轮次")
111        rt.step(n=3)
112        out = rt.emit()
113
114        # 每轮 emit 后只有 1 个 pending signal
115        assert len(rt.state.pending_signals) == 1, \
116            f"Round {round_i}: expected 1 pending after emit, got {len(rt.state.pending_signals)}"
117
118        total_mu = sum(rt.state.mu.values())
119        mu_deltas.append(total_mu)
120
121    # mu 不应无界增长（如果有双重应用会爆炸）
122    assert mu_deltas[-1] < mu_deltas[0] * 20, \
123        f"mu growing too fast, possible double application: {mu_deltas}"
124
125    print(f"  PASS: no double application over 5 rounds, mu_deltas={[round(d,2) for d in mu_deltas]}")
126
127
128def run_all():
129    tests = [
130        ("no_feedback_loop_method", test_no_feedback_loop_method),
131        ("pending_signal_sole_entry", test_pending_signal_is_sole_entry),
132        ("emit_feedback_exactly_once", test_emit_feedback_exactly_once),
133        ("commit_feedback_via_signal", test_commit_feedback_via_signal_only),
134        ("no_double_application", test_no_double_application),
135    ]
136
137    passed = 0
138    failed = 0
139    for name, fn in tests:
140        try:
141            print(f"[EXACTLY-ONCE] {name}...")
142            fn()
143            passed += 1
144        except Exception as e:
145            print(f"  FAIL: {e}")
146            failed += 1
147
148    print(f"\n{'='*50}")
149    print(f"Exactly-Once Gate: {passed} passed, {failed} failed, {len(tests)} total")
150    print(f"{'='*50}")
151    return failed == 0
152
153
154if __name__ == '__main__':
155    success = run_all()
156    sys.exit(0 if success else 1)