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