bingshuo/guanghu-ice-heart
Watch
1
0
Fork
You've already forked guanghu-ice-heart
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity

feat: define Guanghu systems channels and persona body

Browse source
This commit is contained in:
冰朔 2026-09-08 16:14:14 +08:00
commit c3f133ecf8
23 changed files with 562 additions and 57 deletions
Download patch file Download diff file Expand all files Collapse all files

25
server-tools/persona-dynamic-subagent-command/test_command_router.py
View file

@ -10,8 +10,10 @@ def event(persona='ICE-P-ZY001', session='s1'):
class Tests(unittest.TestCase):
def test_every_slot_is_model_backed_and_owned_by_parent(self):
with tempfile.TemporaryDirectory() as t, patch.object(M,'runtime_available',return_value=True):
v=M.activate(event(),Path(t)); self.assertEqual(len(v['team']),4)
v=M.activate(event(),Path(t)); self.assertEqual(len(v['team']),8)
self.assertTrue(all(x['model_cognition_required'] and x['controller']=='ICE-P-ZY001' for x in v['team']))
self.assertTrue(all(x['runtime_state']=='DORMANT_READY' and not x['model_context_loaded'] for x in v['team']))
self.assertEqual(v['active_agent_count'],0)
def test_persona_switch_rebinds_all_slots_and_invalidates_generation(self):
personas=iter(sorted(M.registered_personas()))
first=next(personas); second=next(x for x in personas if x!=first)
@ -24,5 +26,26 @@ class Tests(unittest.TestCase):
with tempfile.TemporaryDirectory() as t:
with self.assertRaisesRegex(M.CommandError,'NOT_REGISTERED'): M.activate(event('PER-NOT-REAL'),Path(t))
with patch.object(M,'runtime_available',return_value=False), self.assertRaisesRegex(M.CommandError,'PROVIDER_UNAVAILABLE'): M.activate(event(),Path(t))
def test_dispatch_activates_only_selected_slot_then_releases_it(self):
with tempfile.TemporaryDirectory() as t, patch.object(M,'runtime_available',return_value=True):
state=Path(t); current=M.activate(event(),state); selected=current['team'][0]
def fake(*args,**kwargs):
during=M.load(state/'CURRENT.json')
self.assertEqual(during['active_agent_count'],1)
self.assertEqual([x['slot_id'] for x in during['team'] if x['runtime_state']=='ACTIVE'],[selected['slot_id']])
return __import__('subprocess').CompletedProcess(args[0],0,'{}','')
with patch.object(M.subprocess,'run',side_effect=fake):
receipt=M.dispatch({'generation':current['generation'],'child_agent_id':selected['child_agent_id'],'task':'read only'},state)
after=M.load(state/'CURRENT.json')
self.assertEqual(receipt['final_slot_state'],'DORMANT_READY'); self.assertEqual(after['active_agent_count'],0)
def test_failed_limb_emits_symptom_for_treatment_team(self):
with tempfile.TemporaryDirectory() as t, patch.object(M,'runtime_available',return_value=True):
state=Path(t); current=M.activate(event(),state); selected=current['team'][0]
failed=__import__('subprocess').CompletedProcess([],1,'','broken')
with patch.object(M.subprocess,'run',return_value=failed):
receipt=M.dispatch({'generation':current['generation'],'child_agent_id':selected['child_agent_id'],'task':'read only'},state)
after=M.load(state/'CURRENT.json')
self.assertEqual(receipt['final_slot_state'],'SYMPTOM_REPORTED')
self.assertEqual(after['treatment_signal']['treatment_router'],'TCS-AGENT-TREATMENT-ROUTER-001')
if __name__=='__main__': unittest.main()