WeiCHEN
47fdeb1655
Replace Linux /proc hard dependency with Darwin ps+sysctl while keeping zombie/group/boot three-value semantics fail-closed for unreadable PIDs.
142 lines
7.8 KiB
Python
142 lines
7.8 KiB
Python
"""Real Git/SQLite tasks with a fake Herdr transport; never starts paid agents."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
import test_agentctl as fixtures
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / 'herdr_agents.py'
|
|
sys.path.insert(0, str(SCRIPT.parent))
|
|
from herdr_agents import PROFILES, ROUTES
|
|
|
|
|
|
class HerdrTests(unittest.TestCase):
|
|
git = fixtures.AgentctlTests.git
|
|
cli = fixtures.AgentctlTests.cli
|
|
initialized_task = fixtures.AgentctlTests.initialized_task
|
|
|
|
def setUp(self):
|
|
fixtures.AgentctlTests.setUp(self)
|
|
self.task = self.initialized_task()
|
|
self.bin = Path(self.temp.name) / 'bin'; self.bin.mkdir()
|
|
self.log = Path(self.temp.name) / 'calls.jsonl'
|
|
# Resolve so macOS /var → /private/var matches agentctl Repo.current strings.
|
|
task_cwd = str(self.task.resolve())
|
|
self.env = dict(os.environ, HERDR_ENV='1', PATH=str(self.bin)+os.pathsep+os.environ['PATH'],
|
|
MOCK_LOG=str(self.log), MOCK_CWD=task_cwd, MOCK_MODE='ok')
|
|
self.env['MOCK_AGENTS'] = json.dumps(['qwen_max__alpha','qwen_flash__alpha','context_kimi__alpha','implementer_cursor__alpha','reviewer_glm__alpha'])
|
|
fake = self.bin / 'herdr'
|
|
fake.write_text("#!/usr/bin/env python3\n" + r'''import json,os,re,sys
|
|
from pathlib import Path
|
|
a=sys.argv[1:]; log=Path(os.environ['MOCK_LOG'])
|
|
with log.open('a') as f:f.write(json.dumps(a)+'\n')
|
|
mode=os.environ['MOCK_MODE']; cwd=os.environ['MOCK_CWD']
|
|
if a[:2]==['agent','list']:
|
|
result={'agents':[{'name':n,'cwd':cwd,'agent_status':'idle'} for n in json.loads(os.environ['MOCK_AGENTS'])]}
|
|
elif a[:2]==['agent','get']:
|
|
result={'agent':{'agent_status':'blocked' if mode=='blocked' else 'done'}}
|
|
elif a[:2]==['pane','list']:
|
|
result={'panes':[{'pane_id':'w1:p2','cwd':cwd}]}
|
|
elif a[:2]==['agent','prompt']:
|
|
if mode=='timeout':print('timeout agent_blocked quota',file=sys.stderr);sys.exit(1)
|
|
result={'status':'done'}
|
|
elif a[:2]==['agent','read']:
|
|
if mode=='retry':
|
|
prompts=[json.loads(l) for l in log.read_text().splitlines() if json.loads(l)[:2]==['agent','prompt']]
|
|
if len(prompts)==1:print(re.search(r'HERDR_RETRY:[a-f0-9]+',prompts[-1][3])[0]+':quota');sys.exit(0)
|
|
print('Finished with evidence');sys.exit(0)
|
|
else:result={'status':'started'}
|
|
print(json.dumps({'result':result}))
|
|
''')
|
|
fake.chmod(0o755)
|
|
|
|
def run_cli(self, *args, env=None, cwd=None):
|
|
return subprocess.run([sys.executable,'-B',str(SCRIPT),*args],cwd=cwd or self.task,
|
|
env=env or self.env,capture_output=True,text=True,timeout=15)
|
|
|
|
def calls(self, group):
|
|
if not self.log.exists():return []
|
|
return [a for a in map(json.loads,self.log.read_text().splitlines()) if a[:2]==['agent',group]]
|
|
|
|
def test_all_profiles_have_noninteractive_flags_and_explicit_pi_routes(self):
|
|
for profile,(name,_,native) in PROFILES.items():
|
|
with self.subTest(profile=profile):
|
|
p=self.run_cli('start','--dry-run',profile,name+'__a','w1:p2',env=dict(self.env,HERDR_ENV=''))
|
|
self.assertEqual(p.returncode,0,p.stderr)
|
|
for arg in native:self.assertIn(arg,p.stdout)
|
|
self.assertFalse(self.log.exists())
|
|
self.assertIn('qwen3.8-max',PROFILES['pi-max'][2])
|
|
self.assertIn('--auto',PROFILES['kimi'][2])
|
|
self.assertIn('danger-full-access',PROFILES['codex'][2])
|
|
|
|
def test_outside_herdr_and_wrong_task_cwd_never_contact_transport(self):
|
|
for env,cwd in [(dict(self.env,HERDR_ENV=''),self.task),(self.env,self.root)]:
|
|
p=self.run_cli('dispatch','--task','alpha','report','draft',env=env,cwd=cwd)
|
|
self.assertNotEqual(p.returncode,0)
|
|
self.assertFalse(self.log.exists())
|
|
|
|
def test_main_or_wrong_target_pane_cannot_start(self):
|
|
p=self.run_cli('start','--task','alpha','pi-max','qwen_max__a','w1:p2',env=dict(self.env,MOCK_CWD=str(self.root)))
|
|
self.assertNotEqual(p.returncode,0);self.assertFalse(self.calls('start'))
|
|
p=self.run_cli('start','--task','alpha','pi-max','qwen_max__a','w1:p2')
|
|
self.assertEqual(p.returncode,0,p.stderr);self.assertEqual(len(self.calls('start')),1)
|
|
|
|
def test_report_routes_to_max_not_kimi_or_minimax(self):
|
|
p=self.run_cli('dispatch','--task','alpha','report','draft')
|
|
self.assertEqual(p.returncode,0,p.stderr)
|
|
self.assertEqual(self.calls('prompt')[0][2],'qwen_max__alpha')
|
|
self.assertTrue(all('context_kimi' not in v for k,v in ROUTES.items() if k not in ('report-final-check','quota-emergency')))
|
|
|
|
def test_implementation_fallback_is_explicit(self):
|
|
self.env['MOCK_AGENTS']=json.dumps(['reviewer_glm__alpha'])
|
|
p=self.run_cli('dispatch','--task','alpha','implementation','change')
|
|
self.assertEqual(p.returncode,69);self.assertFalse(self.calls('prompt'))
|
|
p=self.run_cli('dispatch','--task','alpha','--implementation-fallback','Cursor quota exhausted, no writes','implementation','change')
|
|
self.assertEqual(p.returncode,0,p.stderr)
|
|
self.assertEqual(self.calls('prompt')[0][2],'reviewer_glm__alpha')
|
|
|
|
def test_uncertain_transport_error_and_blocked_do_not_fallback(self):
|
|
for mode in ['timeout','blocked']:
|
|
self.log.unlink(missing_ok=True)
|
|
p=self.run_cli('dispatch','--task','alpha','report','draft',env=dict(self.env,MOCK_MODE=mode))
|
|
self.assertEqual(p.returncode,75,p.stderr);self.assertEqual(len(self.calls('prompt')),1)
|
|
|
|
def test_explicit_prewrite_retry_reaches_flash(self):
|
|
p=self.run_cli('dispatch','--task','alpha','report','draft',env=dict(self.env,MOCK_MODE='retry'))
|
|
self.assertEqual(p.returncode,0,p.stderr)
|
|
self.assertEqual([a[2] for a in self.calls('prompt')],['qwen_max__alpha','qwen_flash__alpha'])
|
|
|
|
def test_same_role_ambiguity_and_foreign_cwd_do_not_dispatch(self):
|
|
for env in [dict(self.env,MOCK_AGENTS=json.dumps(['qwen_max__a','qwen_max__b'])),dict(self.env,MOCK_CWD=str(self.root))]:
|
|
p=self.run_cli('dispatch','--task','alpha','report','draft',env=env)
|
|
self.assertNotEqual(p.returncode,0);self.assertFalse(self.calls('prompt'))
|
|
|
|
def test_kimi_reason_disabled_marker_and_cross_lane_budget(self):
|
|
p=self.run_cli('dispatch','--task','alpha','report-final-check','check')
|
|
self.assertNotEqual(p.returncode,0);self.assertFalse(self.calls('prompt'))
|
|
marker=self.root/'.locks/herdr-kimi.disabled';marker.touch()
|
|
args=['dispatch','--task','alpha','--kimi-reason','final report','report-final-check','check']
|
|
self.assertNotEqual(self.run_cli(*args).returncode,0);self.assertFalse(self.calls('prompt'))
|
|
marker.unlink()
|
|
p=self.run_cli(*args);self.assertEqual(p.returncode,0,p.stderr)
|
|
p=self.run_cli('dispatch','--task','alpha','--kimi-reason','outage','quota-emergency','check')
|
|
self.assertNotEqual(p.returncode,0);self.assertEqual(len(self.calls('prompt')),1)
|
|
|
|
def test_kimi_atomic_limit_across_eight_processes(self):
|
|
args=[sys.executable,'-B',str(SCRIPT),'dispatch','--task','alpha','--kimi-reason','final report','report-final-check','check']
|
|
processes=[subprocess.Popen(args,cwd=self.task,env=self.env,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True) for _ in range(8)]
|
|
for p in processes:p.communicate(timeout=20)
|
|
self.assertEqual(sum(p.returncode==0 for p in processes),1)
|
|
self.assertEqual(len(self.calls('prompt')),1)
|
|
|
|
def test_kimi_uncertain_send_consumes_budget(self):
|
|
args=['dispatch','--task','alpha','--kimi-reason','final report','report-final-check','check']
|
|
self.assertEqual(self.run_cli(*args,env=dict(self.env,MOCK_MODE='timeout')).returncode,75)
|
|
self.assertNotEqual(self.run_cli(*args).returncode,0)
|
|
self.assertEqual(len(self.calls('prompt')),1)
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|