weichen
8bc7940ee0
Align prepared clients with role budgets and automatic approvals. Keep uncertain delivery from triggering duplicate work and preserve task worktree gates.
140 lines
7.7 KiB
Python
140 lines
7.7 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'
|
|
self.env = dict(os.environ, HERDR_ENV='1', PATH=str(self.bin)+os.pathsep+os.environ['PATH'],
|
|
MOCK_LOG=str(self.log), MOCK_CWD=str(self.task), 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()
|