CIE-Unified

test_smoke.py

  1"""
  2CIE Smoke Tests (SPEC §7.1)
  3
  41. 冷启动能否点火
  52. output-to-input 回灌是否真实存在
  63. 无任务时是否出现归巢
  74. 衰减/遗忘是否发生
  85. 任务切换时激活核是否迁移
  96. 资源不足时是否允许降级输出
 10"""
 11
 12import sys
 13import os
 14sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 15
 16from cie.runtime import CIERuntime
 17
 18
 19def test_01_cold_start():
 20    """冷启动能否点火"""
 21    rt = CIERuntime(seed=42)
 22    assert rt.graph.node_count == 0
 23    assert rt.state.step_count == 0
 24
 25    # 注入输入（PendingSignal 排队，step 时消费）
 26    rt.ingest("你好")
 27    rt.step(3)
 28    assert rt.graph.node_count >= 2, f"Expected >=2 nodes, got {rt.graph.node_count}"
 29    assert rt.state.step_count == 3
 30
 31    # 能产出输出
 32    output = rt.emit()
 33    assert output is not None
 34    assert 'mode' in output
 35    assert 'activated' in output
 36    print(f"  PASS: cold start -> {rt.graph.node_count} nodes, "
 37          f"{len(rt.state.active_region)} active, mode={output['mode']}")
 38
 39
 40def test_02_output_to_input_feedback():
 41    """output-to-input 回灌是否真实存在"""
 42    rt = CIERuntime(seed=42)
 43
 44    # 第一轮
 45    rt.ingest("你好")
 46    rt.step(3)
 47    out1 = rt.emit()
 48
 49    # 记录第一轮后的状态
 50    phi_before = dict(rt.state.phi)
 51    mu_before = dict(rt.state.mu)
 52
 53    # 第二轮——PendingSignal 在 step 时消费
 54    rt.ingest("世界")
 55    rt.step(1)  # 消费 emit 回灌 + ingest 信号
 56    feedback_happened = False
 57    if out1['activated']:
 58        for item in out1['activated']:
 59            nid = item['node']
 60            if (rt.state.phi.get(nid, 0.0) != phi_before.get(nid, 0.0) or
 61                rt.state.mu.get(nid, 0.0) != mu_before.get(nid, 0.0)):
 62                feedback_happened = True
 63                break
 64
 65    assert feedback_happened, "Output-to-input feedback did not happen"
 66    print(f"  PASS: output-to-input feedback confirmed")
 67
 68
 69def test_03_homing_without_task():
 70    """无任务时是否出现归巢"""
 71    rt = CIERuntime(seed=42)
 72
 73    # 建立一些结构
 74    rt.ingest("学习知识", anchors=["学习"])
 75    rt.step(10)
 76
 77    # 记录锚点
 78    anchors_before = set(rt.state.anchor_nodes)
 79
 80    # 不再注入新任务，继续跑
 81    initial_active = set(rt.state.active_region)
 82    rt.step(20)
 83    later_active = set(rt.state.active_region)
 84
 85    # 活跃区域应该收缩（归巢=激活核回落）
 86    # 或者 mu 总量下降
 87    mu_sum_before = sum(rt.state.mu.get(n, 0.0) for n in initial_active)
 88    mu_sum_after = sum(rt.state.mu.get(n, 0.0) for n in later_active)
 89
 90    # 归巢的证据：活跃区域缩小或激活降低
 91    shrunk = len(later_active) <= len(initial_active)
 92    decayed = mu_sum_after <= mu_sum_before + 1e-6  # allow tiny float error
 93
 94    assert shrunk or decayed, (
 95        f"No homing: active {len(initial_active)}->{len(later_active)}, "
 96        f"mu_sum {mu_sum_before:.3f}->{mu_sum_after:.3f}"
 97    )
 98    print(f"  PASS: homing observed — active {len(initial_active)}->{len(later_active)}, "
 99          f"mu_sum {mu_sum_before:.3f}->{mu_sum_after:.3f}")
100
101
102def test_04_decay_and_forgetting():
103    """衰减/遗忘是否发生"""
104    rt = CIERuntime(seed=42)
105
106    rt.ingest("记忆衰减测试")
107    rt.step(5)
108
109    # 记录当前 phi
110    phi_snapshot = {k: v for k, v in rt.state.phi.items() if abs(v) > 1e-10}
111
112    # 继续跑很多步（无新输入）
113    rt.step(50)
114
115    # 检查衰减
116    decayed_count = 0
117    for nid, old_phi in phi_snapshot.items():
118        new_phi = rt.state.phi.get(nid, 0.0)
119        if abs(new_phi) < abs(old_phi) - 1e-10:
120            decayed_count += 1
121
122    # 检查 decay_events 是否有记录
123    has_decay_events = len(rt.state.decay_events) > 0
124
125    assert decayed_count > 0 or has_decay_events, "No decay/forgetting observed"
126    print(f"  PASS: decay confirmed — {decayed_count} nodes decayed, "
127          f"{len(rt.state.decay_events)} decay events")
128
129
130def test_05_task_switch_activation_migrates():
131    """任务切换时激活核是否迁移"""
132    rt = CIERuntime(seed=42)
133
134    # 任务 1
135    rt.ingest("数学计算")
136    rt.step(5)
137    active_task1 = set(rt.state.active_region)
138
139    # 任务 2（不同领域）
140    rt.ingest("音乐欣赏")
141    rt.step(5)
142    active_task2 = set(rt.state.active_region)
143
144    # 激活区域应该迁移——新任务的节点应该出现
145    new_nodes = active_task2 - active_task1
146    assert len(new_nodes) > 0, "Activation did not migrate to new task"
147
148    # 旧任务的节点应该有所减弱
149    old_only = active_task1 - active_task2
150    # 至少新节点出现了就说明迁移发生
151    print(f"  PASS: activation migrated — {len(new_nodes)} new nodes, "
152          f"{len(old_only)} old-only nodes")
153
154
155def test_06_degraded_output():
156    """资源不足时是否允许降级输出"""
157    rt = CIERuntime(seed=42)
158
159    # 最小输入
160    rt.ingest("嗯")
161    # 不 step，直接 emit
162    output = rt.emit()
163
164    # 或者：消耗大量注意力后再试
165    rt2 = CIERuntime(seed=42)
166    # 大量输入占满注意力
167    rt2.ingest("这是一段很长的测试文本用来占满注意力池的容量看看会不会降级输出")
168    rt2.step(2)
169    out2 = rt2.emit()
170
171    # 至少有一个应该是降级的
172    modes = {output['mode'], out2['mode']}
173    has_non_full = 'degraded' in modes or 'minimal' in modes
174
175    # 即使是 full，只要能输出就算通过（降级是允许不完整，不是强制不完整）
176    assert output is not None and out2 is not None, "Output is None"
177    print(f"  PASS: output modes = {modes}, both produced output")
178
179
180# ── 运行所有测试 ──
181
182if __name__ == '__main__':
183    tests = [
184        test_01_cold_start,
185        test_02_output_to_input_feedback,
186        test_03_homing_without_task,
187        test_04_decay_and_forgetting,
188        test_05_task_switch_activation_migrates,
189        test_06_degraded_output,
190    ]
191
192    passed = 0
193    failed = 0
194    for t in tests:
195        name = t.__doc__.strip() if t.__doc__ else t.__name__
196        try:
197            print(f"[SMOKE] {name}")
198            t()
199            passed += 1
200        except Exception as e:
201            print(f"  FAIL: {e}")
202            failed += 1
203
204    print(f"\n{'='*50}")
205    print(f"Smoke Tests: {passed} passed, {failed} failed, {passed+failed} total")
206    print(f"{'='*50}")